"""Connectors domain — business logic.""" from __future__ import annotations from typing import Any, Awaitable, Callable, Optional from uuid import UUID import structlog from src.core.exceptions import ConflictError, NotFoundError, ValidationError from src.core.security.cloud_tokens import ( create_connector_approval_token, decode_connector_approval_token, ) from src.core.security.encryption import decrypt_value, encrypt_value from src.domain.connectors.credential_check import credential_check_for, verify_connector_credentials from src.domain.connectors.cloud_status import cloud_connector_runtime_status from src.domain.connectors.models import ( ConnectorConnection, ConnectorPermission, WorkspaceConnectorDefault, ) from src.domain.connectors.providers import ( CONNECTORS, ConnectorActionDef, ConnectorDef, get_action, get_connector, ) from src.domain.connectors.repository import ConnectorRepository, WorkspaceConnectorDefaultRepository from src.domain.connectors.schemas import ( ConnectorActionResponse, ConnectorApprovalResponse, ConnectorChangelogEntryResponse, ConnectorConnectionResponse, ConnectorFieldResponse, ConnectorPermissionResponse, ConnectorResponse, ConnectorTriggerResponse, CopyFromWorkspaceDefaultRequest, CreateConnectorApprovalRequest, CreateConnectorConnectionRequest, RotateConnectorSecretsRequest, UpdateConnectorConnectionRequest, UpsertConnectorPermissionRequest, UpsertWorkspaceConnectorDefaultRequest, WorkspaceConnectorDefaultResponse, ) from src.integrations.cloud_control_client import CloudControlClient logger = structlog.get_logger() EMAIL_CONNECTOR_KEYS = ("resend", "mailgun", "ses") def _remember_outbound_crm_mutation(plan: Any, capability: str, payload: dict[str, Any]) -> None: from src.domain.connectors.crm_events import record_crm_outbound_mutation parts = capability.split(".") if len(parts) < 3: return object_type = parts[1] verb = parts[2] if object_type not in {"contact", "company", "deal"} or verb not in { "create", "update", "upsert", }: return blob = payload.get(object_type) if isinstance(payload.get(object_type), dict) else {} ref = blob.get("ref") if isinstance(blob.get("ref"), dict) else {} external_id = ref.get("external_id") fields = { key for key, value in blob.items() if key not in {"ref", "properties"} and value is not None } record_crm_outbound_mutation( connection_id=plan.connection_id, object_type=object_type, external_id=str(external_id) if external_id else None, fields=fields, ) def _cloud_permission_payloads( *, connection_id: UUID, project_id: Optional[UUID], permissions: list[ConnectorPermission], ) -> list[dict[str, Any]]: """Build permission rows for Cloud control sync (requires connection_id). Deny modes are omitted — Cloud treats missing rows as deny. """ cid = str(connection_id) pid = str(project_id) if project_id else None return [ { "connection_id": cid, "project_id": pid, "action_slug": p.action_slug, "mode": p.mode, "policy": p.policy_json or {}, } for p in permissions if p.mode != "deny" ] class ConnectorExecutionDecision: def __init__( self, *, allowed: bool, reason_code: Optional[str] = None, message: Optional[str] = None, connection: Optional[ConnectorConnection] = None, connector: Optional[ConnectorDef] = None, action: Optional[ConnectorActionDef] = None, permission: Optional[ConnectorPermission] = None, ) -> None: self.allowed = allowed self.reason_code = reason_code self.message = message self.connection = connection self.connector = connector self.action = action self.permission = permission class ConnectorService: def __init__( self, *, repo: ConnectorRepository, cloud_client: CloudControlClient, ) -> None: self._repo = repo self._cloud = cloud_client async def _sync_connector_catalog_entry(self, cdef: ConnectorDef) -> UUID: """Persist one catalog connector and its actions/triggers. Returns connector PK.""" row = await self._repo.upsert_catalog_connector( cdef.key, name=cdef.name, category=cdef.category, auth_type=cdef.auth_type, status=cdef.status, icon=cdef.icon, description=cdef.description, docs_url=cdef.docs_url, metadata_json={ "config_fields": list(cdef.config_fields), "secret_fields": list(cdef.secret_fields), "version": cdef.version, "image_url": cdef.image_url, "client_side": cdef.client_side, "ownership_mode": cdef.ownership_mode, "execution_domain": cdef.execution_domain, "changelog": [ {"version": e.version, "date": e.date, "notes": e.notes} for e in cdef.changelog ], }, ) # Capture before any further await — async ORM cannot lazy-load row.id later. connector_id = row.id for action in cdef.actions: await self._repo.upsert_catalog_action( connector_id, action.key, slug=action.slug, name=action.name, description=action.description, input_schema=action.input_schema, output_schema=action.output_schema, execution_mode=action.execution_mode, required_scopes=list(action.required_scopes), required_capabilities=list(action.required_capabilities), ) for trigger in cdef.triggers: await self._repo.upsert_catalog_trigger( connector_id, trigger.key, event_type=trigger.event_type, name=trigger.name, webhook_required=trigger.webhook_required, polling_supported=trigger.polling_supported, schema=trigger.schema, ) return connector_id async def sync_catalog_defaults(self) -> None: """Persist the static catalog so DB-backed consumers can query it.""" for cdef in CONNECTORS.values(): await self._sync_connector_catalog_entry(cdef) async def list_catalog(self) -> list[ConnectorResponse]: return [_connector_response(cdef) for cdef in CONNECTORS.values()] async def get_catalog_connector(self, key: str) -> ConnectorResponse: cdef = _require_connector(key) return _connector_response(cdef) async def list_connections(self, project_id: UUID) -> list[ConnectorConnectionResponse]: rows = await self._repo.list_connections_for_project(project_id) return [await self._connection_response(row) for row in rows] async def get_connection( self, project_id: UUID, connection_id: UUID ) -> ConnectorConnectionResponse: row = await self._require_project_connection(project_id, connection_id) return await self._connection_response(row) async def create_connection( self, *, workspace_id: UUID, project_id: UUID, body: CreateConnectorConnectionRequest, ) -> ConnectorConnectionResponse: cdef = _require_connector(body.connector_key) catalog_connector_id = await self._sync_connector_catalog_entry(cdef) if body.provider_connection_id: from src.domain.provider_auth.facades import execution_credentials if set(body.secrets) & {"access_token", "refresh_token", "client_id", "client_secret"} or body.scopes: raise ValidationError("Platform OAuth scopes and credentials are managed by Connect") credentials, scopes = await execution_credentials(self._repo._db, workspace_id=workspace_id, project_id=project_id, connection_id=body.provider_connection_id, connector_key=cdef.key) config = dict(body.config) if cdef.key == "microsoft_ads" and not config.get("client_id"): from src.config import get_settings config["client_id"] = get_settings().microsoft_ads_platform_client_id for field in ("instance_url", "portal_id"): if field not in config and credentials.get(field): config[field] = credentials[field] await self._validate_config_and_secrets( cdef, config, body.secrets, require_all_secrets=False, platform_grant=True ) if cdef.key == "google_ads" and not body.secrets.get("developer_token", "").strip(): raise ValidationError("Google Ads requires your developer token") if cdef.key == "microsoft_ads" and not body.secrets.get("developer_token", "").strip(): raise ValidationError("Microsoft Ads requires your developer token") merged_secrets = { key: value for key, value in {**body.secrets, **credentials}.items() if key in cdef.secret_fields and value } await verify_connector_credentials(cdef.key, config, merged_secrets) if await self._repo.get_connection_by_name(project_id, body.name): raise ConflictError(f"Connector connection {body.name!r} already exists") row = await self._repo.create_connection( connector_id=catalog_connector_id, workspace_id=workspace_id, project_id=project_id, name=body.name, status="active", scopes=scopes, config_json=config, provider_connection_id=body.provider_connection_id, secrets_enc={k: encrypt_value(v) for k, v in merged_secrets.items()}) row = await self._repo.update_connection(row, auth_data_ref=_secret_ref(row)) for action in cdef.actions: await self._repo.upsert_permission(row.id, action.slug, mode=_permission_mode_for_create(cdef, body.permissions, action.slug)) await self._push_connection_to_cloud(row, cdef, push_credentials=True) return await self._connection_response(row) await self._validate_config_and_secrets( cdef, body.config, body.secrets, require_all_secrets=True ) await verify_connector_credentials(cdef.key, body.config, body.secrets) existing = await self._repo.get_connection_by_name(project_id, body.name) if existing: raise ConflictError(f"Connector connection {body.name!r} already exists") secrets_enc = {key: encrypt_value(val) for key, val in body.secrets.items()} row = await self._repo.create_connection( connector_id=catalog_connector_id, workspace_id=workspace_id, project_id=project_id, name=body.name, status="active", auth_data_ref=None, scopes=body.scopes, config_json=body.config, secrets_enc=secrets_enc or None, ) connection_id = row.id auth_data_ref = f"connector:{workspace_id}:{project_id}:{connection_id}" row = await self._repo.update_connection(row, auth_data_ref=auth_data_ref) for action in cdef.actions: mode = _permission_mode_for_create(cdef, body.permissions, action.slug) await self._repo.upsert_permission(connection_id, action.slug, mode=mode) await self._push_connection_to_cloud(row, cdef, push_credentials=True) logger.info("connector_connection_created", project_id=str(project_id), connector=cdef.key) row = await self._require_project_connection(project_id, connection_id) return await self._connection_response(row) async def update_connection( self, project_id: UUID, connection_id: UUID, body: UpdateConnectorConnectionRequest, ) -> ConnectorConnectionResponse: row = await self._require_project_connection(project_id, connection_id) cdef = await self._connector_def_for_row(row) fields: dict[str, Any] = {} if body.name is not None: existing = await self._repo.get_connection_by_name(project_id, body.name) if existing and existing.id != row.id: raise ConflictError(f"Connector connection {body.name!r} already exists") fields["name"] = body.name if body.status is not None: fields["status"] = body.status if body.config is not None: config = dict(body.config) if row.provider_connection_id: for key in _PLATFORM_GRANT_CONFIG: if key not in config and (row.config_json or {}).get(key): config[key] = row.config_json[key] stored_secrets = _decrypt_secret_map(row.secrets_enc) await self._validate_config_and_secrets( cdef, config, {}, platform_grant=bool(row.provider_connection_id) ) await verify_connector_credentials(cdef.key, config, stored_secrets) fields["config_json"] = config if body.scopes is not None: if row.provider_connection_id: raise ValidationError("Platform Google scopes are managed by Connect") fields["scopes"] = body.scopes if fields: row = await self._repo.update_connection(row, **fields) await self._push_connection_to_cloud(row, cdef) return await self._connection_response(row) async def rotate_secrets( self, project_id: UUID, connection_id: UUID, body: RotateConnectorSecretsRequest, ) -> ConnectorConnectionResponse: row = await self._require_project_connection(project_id, connection_id) cdef = await self._connector_def_for_row(row) if row.provider_connection_id: raise ValidationError("Platform OAuth credentials are managed by Connect") await self._validate_config_and_secrets(cdef, row.config_json or {}, body.secrets) merged_plain = { key: value for key, value in {**_decrypt_secret_map(row.secrets_enc), **body.secrets}.items() if key in cdef.secret_fields and str(value).strip() } await verify_connector_credentials(cdef.key, row.config_json or {}, merged_plain) merged = {key: encrypt_value(val) for key, val in merged_plain.items() if str(val).strip()} auth_data_ref = f"connector:{row.workspace_id}:{row.project_id}:{row.id}" row = await self._repo.update_connection( row, secrets_enc=merged, auth_data_ref=auth_data_ref ) await self._push_connection_to_cloud(row, cdef, push_credentials=True) logger.info("connector_connection_secrets_rotated", connection_id=str(connection_id)) row = await self._require_project_connection(project_id, connection_id) return await self._connection_response(row) async def delete_connection(self, project_id: UUID, connection_id: UUID) -> None: row = await self._require_project_connection(project_id, connection_id) await self._cloud.delete_connector_connection(str(row.id)) await self._repo.delete_connection(row) async def allow_all_actions( self, project_id: UUID, connection_id: UUID ) -> ConnectorConnectionResponse: """Set allow on every action for this connection (syncs each permission to Cloud).""" row = await self._require_project_connection(project_id, connection_id) cdef = await self._connector_def_for_row(row) conn_id = row.id pid = str(row.project_id) if row.project_id else None for action in cdef.actions: permission = await self._repo.upsert_permission( conn_id, action.slug, mode="allow" ) await self._cloud.upsert_connector_permission( connection_id=str(conn_id), project_id=pid, action_slug=permission.action_slug, mode=permission.mode, policy=permission.policy_json or {}, ) row = await self._require_project_connection(project_id, connection_id) return await self._connection_response(row) async def upsert_permission( self, project_id: UUID, connection_id: UUID, body: UpsertConnectorPermissionRequest, ) -> ConnectorPermissionResponse: row = await self._require_project_connection(project_id, connection_id) cdef = await self._connector_def_for_row(row) if not any(action.slug == body.action_slug for action in cdef.actions): raise ValidationError(f"Action {body.action_slug!r} does not belong to connector {cdef.key!r}") permission = await self._repo.upsert_permission( row.id, body.action_slug, mode=body.mode, policy_json=body.policy, ) await self._cloud.upsert_connector_permission( connection_id=str(row.id), project_id=str(row.project_id) if row.project_id else None, action_slug=permission.action_slug, mode=permission.mode, policy=permission.policy_json or {}, ) return _permission_response(permission) async def create_approval( self, *, workspace_id: UUID, project_id: UUID, connection_id: UUID, user_id: UUID, body: CreateConnectorApprovalRequest, ) -> ConnectorApprovalResponse: row = await self._require_project_connection(project_id, connection_id) if row.workspace_id != workspace_id: raise NotFoundError("Connector connection not found") cdef = await self._connector_def_for_row(row) action_pair = get_action(body.action_slug) if not action_pair or action_pair[0].key != cdef.key: raise ValidationError( f"Action {body.action_slug!r} does not belong to connector {cdef.key!r}" ) permission = await self._repo.get_permission(row.id, body.action_slug) if not permission or permission.mode != "ask": raise ValidationError("Connector action does not require approval") token, expires_at = create_connector_approval_token( workspace_id=workspace_id, project_id=project_id, connection_id=connection_id, action_slug=body.action_slug, user_id=user_id, run_id=body.run_id, ttl_minutes=body.ttl_minutes, ) return ConnectorApprovalResponse(approval_token=token, expires_at=expires_at) async def infer_email_send_action( self, *, project_id: UUID, connection_id: UUID ) -> Optional[str]: """Return ``{key}.send_email`` when the connection is an email provider.""" row = await self._repo.get_connection_for_project(project_id, connection_id) if not row: return None cdef = await self._connector_def_for_row(row) if cdef.category != "email" or cdef.key not in EMAIL_CONNECTOR_KEYS: return None return f"{cdef.key}.send_email" async def resolve_email_send( self, *, workspace_id: UUID, project_id: UUID, enabled_capabilities: Optional[set[str]] = None, provider_hint: Optional[str] = None, approval_token: Optional[str] = None, ) -> Optional[ConnectorExecutionDecision]: """Pick an allowed BYOK email send connection, or None to use platform default.""" rows = await self._repo.list_active_email_connections( project_id, connector_keys=EMAIL_CONNECTOR_KEYS ) hint = (provider_hint or "").strip().lower() if hint in EMAIL_CONNECTOR_KEYS: rows = [pair for pair in rows if pair[1].key == hint] + [ pair for pair in rows if pair[1].key != hint ] for connection, catalog in rows: cdef = get_connector(catalog.key) if cdef and cdef.secret_fields and not connection.secrets_enc: continue decision = await self.authorize_execution( workspace_id=workspace_id, project_id=project_id, connection_id=connection.id, action_slug=f"{catalog.key}.send_email", enabled_capabilities=enabled_capabilities, approval_token=approval_token, ) if decision.allowed: return decision return None async def invoke_messaging_send( self, *, workspace_id: UUID, project_id: UUID, payload: dict[str, Any], enabled_capabilities: Optional[set[str]] = None, provider_hint: Optional[str] = None, binding_connection_id: Optional[UUID] = None, destination_map: Optional[dict[str, str]] = None, approval_token: Optional[str] = None, enforce_invocable: bool = True, enqueue_execute: Optional[ Callable[..., Awaitable[dict[str, Any]]] ] = None, idempotency_key: Optional[str] = None, ): """Resolve messaging.send, re-authorize, then enqueue via integration.execute. Returns a capability-shaped public result. The execution plan stays internal. """ from src.domain.connectors.messaging_resolve import ( DispatchAuthorization, MessagingConnectionSnapshot, dispatch_messaging_send_plan, resolve_messaging_send, to_public_messaging_send_result, ) snapshots: list[MessagingConnectionSnapshot] = [] for row in await self._repo.list_connections_for_project(project_id): if row.workspace_id != workspace_id: continue catalog = await self._repo.get_connector_by_id(row.connector_id) cdef = get_connector(catalog.key) if catalog else None action_slug = "slack.post_message" permission = await self._repo.get_permission(row.id, action_slug) snapshots.append( MessagingConnectionSnapshot( connection_id=row.id, connector_key=catalog.key if catalog else "", status=row.status, has_secrets=bool(row.secrets_enc), permission_mode=permission.mode if permission else "deny", config=row.config_json or {}, requires_secrets=bool(cdef and cdef.secret_fields), client_side_only=bool(cdef and cdef.client_side and not cdef.actions), action_slug=action_slug, ) ) thread_ref = payload.get("thread_ref") if isinstance(payload.get("thread_ref"), str) else None resolution = resolve_messaging_send( payload=payload, connections=snapshots, enabled_capabilities=enabled_capabilities, provider_hint=provider_hint, binding_connection_id=binding_connection_id, destination_map=destination_map, approval_present=bool(approval_token), enforce_invocable=enforce_invocable, ) if not resolution.ok or resolution.plan is None: return to_public_messaging_send_result(resolution, thread_ref=thread_ref) decision = await self.authorize_execution( workspace_id=workspace_id, project_id=project_id, connection_id=resolution.plan.connection_id, action_slug=resolution.plan.action, enabled_capabilities=enabled_capabilities, approval_token=approval_token, ) dispatched = dispatch_messaging_send_plan( resolution.plan, DispatchAuthorization( allowed=decision.allowed, reason_code=decision.reason_code, message=decision.message, ), diagnostics=resolution.diagnostics, ) if not dispatched.ok or dispatched.plan is None: return to_public_messaging_send_result(dispatched, thread_ref=thread_ref) enqueue: dict[str, Any] | None = None if enqueue_execute is not None: plan = dispatched.plan enqueue = await enqueue_execute( connection_id=plan.connection_id, action=plan.action, input=dict(plan.resolved_input), provider_hint=plan.provider, idempotency_key=idempotency_key, approval_token=approval_token, ) return to_public_messaging_send_result( dispatched, thread_ref=thread_ref, enqueue=enqueue ) async def invoke_notification_sms( self, *, workspace_id: UUID, project_id: UUID, payload: dict[str, Any], enabled_capabilities: Optional[set[str]] = None, provider_hint: str | None = None, binding_connection_id: UUID | None = None, approval_token: str | None = None, enqueue_execute: Optional[Callable[..., Awaitable[dict[str, Any]]]] = None, idempotency_key: str | None = None, ): """Resolve and enqueue user-owned Twilio SMS without exposing provider fields.""" from src.domain.connectors.notification_sms import ( NotificationSmsConnectionSnapshot, resolve_notification_sms, to_public_notification_sms_result, ) snapshots = [] for row in await self._repo.list_connections_for_project(project_id): if row.workspace_id != workspace_id: continue catalog = await self._repo.get_connector_by_id(row.connector_id) cdef = get_connector(catalog.key) if catalog else None if not catalog or catalog.key != "twilio": continue permission = await self._repo.get_permission(row.id, "twilio.send_sms") requires_secrets = bool(cdef and cdef.secret_fields) snapshots.append(NotificationSmsConnectionSnapshot( connection_id=row.id, connector_key=catalog.key, status=row.status, has_secrets=bool(row.secrets_enc), permission_mode=permission.mode if permission else "deny", config=row.config_json or {}, requires_secrets=requires_secrets, client_side_only=bool(cdef and cdef.client_side and not cdef.actions), )) resolution = resolve_notification_sms( payload=payload, connections=snapshots, enabled_capabilities=enabled_capabilities, provider_hint=provider_hint, binding_connection_id=binding_connection_id, approval_present=bool(approval_token), enforce_invocable=True, ) if not resolution.ok or resolution.plan is None: return to_public_notification_sms_result(resolution) decision = await self.authorize_execution( workspace_id=workspace_id, project_id=project_id, connection_id=resolution.plan.connection_id, action_slug=resolution.plan.action, enabled_capabilities=enabled_capabilities, approval_token=approval_token, ) if not decision.allowed: from src.domain.connectors.notification_sms import NotificationSmsResolution denied = NotificationSmsResolution( ok=False, reason_code=decision.reason_code or "connector_action_denied", message=decision.message or "Connector action is not allowed", ) return to_public_notification_sms_result(denied) enqueue = None if enqueue_execute is not None: plan = resolution.plan enqueue = await enqueue_execute( connection_id=plan.connection_id, action=plan.action, input=dict(plan.resolved_input), provider_hint=plan.provider, idempotency_key=idempotency_key, approval_token=approval_token, ) return to_public_notification_sms_result(resolution, enqueue=enqueue) async def invoke_automation_workflow_trigger( self, *, workspace_id: UUID, project_id: UUID, payload: dict[str, Any], enabled_capabilities: Optional[set[str]] = None, provider_hint: Optional[str] = None, binding_connection_id: Optional[UUID] = None, workflow_map: Optional[dict[str, UUID]] = None, approval_token: Optional[str] = None, enforce_invocable: bool = True, enqueue_execute: Optional[ Callable[..., Awaitable[dict[str, Any]]] ] = None, idempotency_key: Optional[str] = None, ) -> dict[str, Any]: """Resolve automation.workflow.trigger, re-authorize, enqueue via integration.execute.""" from src.domain.connectors.automation_workflow import ( AutomationWorkflowConnectionSnapshot, DispatchAuthorization, ACTION_BY_CONNECTOR, dispatch_automation_workflow_plan, resolve_automation_workflow_trigger, to_public_automation_workflow_result, ) snapshots: list[AutomationWorkflowConnectionSnapshot] = [] for row in await self._repo.list_connections_for_project(project_id): if row.workspace_id != workspace_id: continue catalog = await self._repo.get_connector_by_id(row.connector_id) cdef = get_connector(catalog.key) if catalog else None action_slug = ACTION_BY_CONNECTOR.get(catalog.key if catalog else "") if not action_slug: action_slug = "" permission = ( await self._repo.get_permission(row.id, action_slug) if action_slug else None ) requires_secrets = False if cdef and cdef.secret_field_defs: requires_secrets = any(field.required for field in cdef.secret_field_defs) elif cdef: requires_secrets = bool(cdef.secret_fields) snapshots.append( AutomationWorkflowConnectionSnapshot( connection_id=row.id, connector_key=catalog.key if catalog else "", status=row.status, has_secrets=bool(row.secrets_enc), permission_mode=permission.mode if permission else "deny", config=row.config_json or {}, requires_secrets=requires_secrets, client_side_only=bool(cdef and cdef.client_side and not cdef.actions), ) ) resolution = resolve_automation_workflow_trigger( payload=payload, connections=snapshots, enabled_capabilities=enabled_capabilities, provider_hint=provider_hint, binding_connection_id=binding_connection_id, workflow_map=workflow_map, approval_present=bool(approval_token), enforce_invocable=enforce_invocable, ) if not resolution.ok or resolution.plan is None: return to_public_automation_workflow_result(resolution) decision = await self.authorize_execution( workspace_id=workspace_id, project_id=project_id, connection_id=resolution.plan.connection_id, action_slug=resolution.plan.action, enabled_capabilities=enabled_capabilities, approval_token=approval_token, ) dispatched = dispatch_automation_workflow_plan( resolution.plan, DispatchAuthorization( allowed=decision.allowed, reason_code=decision.reason_code, message=decision.message, ), diagnostics=resolution.diagnostics, ) if not dispatched.ok or dispatched.plan is None: return to_public_automation_workflow_result(dispatched) enqueue: dict[str, Any] | None = None if enqueue_execute is not None: plan = dispatched.plan enqueue = await enqueue_execute( connection_id=plan.connection_id, action=plan.action, input=dict(plan.resolved_input), provider_hint=plan.connector_key, idempotency_key=idempotency_key, approval_token=approval_token, ) return to_public_automation_workflow_result(dispatched, enqueue=enqueue) async def invoke_document( self, *, workspace_id: UUID, project_id: UUID, capability: str, payload: dict[str, Any], enabled_capabilities: Optional[set[str]] = None, provider_hint: Optional[str] = None, binding_connection_id: Optional[UUID] = None, document_map: Optional[dict[str, UUID]] = None, approval_token: Optional[str] = None, enforce_invocable: bool = True, enqueue_execute: Optional[ Callable[..., Awaitable[dict[str, Any]]] ] = None, idempotency_key: Optional[str] = None, ): """Resolve document.*, re-authorize, enqueue via integration.execute. Catalog invocable is the public gate; this path enforces it by default. """ from src.domain.connectors.document_resolve import ( ACTION_BY_CAPABILITY, DispatchAuthorization, DocumentConnectionSnapshot, dispatch_document_plan, resolve_document, to_public_document_result, ) planned_action = ACTION_BY_CAPABILITY.get(capability, "") snapshots: list[DocumentConnectionSnapshot] = [] for row in await self._repo.list_connections_for_project(project_id): if row.workspace_id != workspace_id: continue catalog = await self._repo.get_connector_by_id(row.connector_id) cdef = get_connector(catalog.key) if catalog else None permission = ( await self._repo.get_permission(row.id, planned_action) if planned_action else None ) snapshots.append( DocumentConnectionSnapshot( connection_id=row.id, connector_key=catalog.key if catalog else "", status=row.status, has_secrets=bool(row.secrets_enc), permission_mode=permission.mode if permission else "deny", config=row.config_json or {}, requires_secrets=bool(cdef and cdef.secret_fields), client_side_only=bool(cdef and cdef.client_side and not cdef.actions), ) ) resolution = resolve_document( capability=capability, payload=payload, connections=snapshots, enabled_capabilities=enabled_capabilities, provider_hint=provider_hint, binding_connection_id=binding_connection_id, document_map=document_map, approval_present=bool(approval_token), enforce_invocable=enforce_invocable, ) if not resolution.ok or resolution.plan is None: return to_public_document_result(resolution, capability=capability) decision = await self.authorize_execution( workspace_id=workspace_id, project_id=project_id, connection_id=resolution.plan.connection_id, action_slug=resolution.plan.action, enabled_capabilities=enabled_capabilities, approval_token=approval_token, ) dispatched = dispatch_document_plan( resolution.plan, DispatchAuthorization( allowed=decision.allowed, reason_code=decision.reason_code, message=decision.message, ), diagnostics=resolution.diagnostics, ) if not dispatched.ok or dispatched.plan is None: return to_public_document_result(dispatched, capability=capability) enqueue: dict[str, Any] | None = None if enqueue_execute is not None: plan = dispatched.plan enqueue = await enqueue_execute( connection_id=plan.connection_id, action=plan.action, input=dict(plan.resolved_input), provider_hint=plan.provider, idempotency_key=idempotency_key, approval_token=approval_token, ) return to_public_document_result( dispatched, capability=capability, enqueue=enqueue ) async def _crm_snapshots( self, *, workspace_id: UUID, project_id: UUID, capability: str, action_by_capability: dict[str, dict[str, str]], ): from src.domain.connectors.crm_resolve import CRMConnectionSnapshot snapshots: list[CRMConnectionSnapshot] = [] for row in await self._repo.list_connections_for_project(project_id): if row.workspace_id != workspace_id: continue catalog = await self._repo.get_connector_by_id(row.connector_id) cdef = get_connector(catalog.key) if catalog else None action = action_by_capability.get(capability, {}).get( catalog.key if catalog else "", "" ) permission = ( await self._repo.get_permission(row.id, action) if action else None ) snapshots.append( CRMConnectionSnapshot( connection_id=row.id, connector_key=catalog.key if catalog else "", status=row.status, has_secrets=bool(row.secrets_enc), permission_mode=permission.mode if permission else "deny", execution_domain="C", requires_secrets=bool(cdef and cdef.secret_fields), client_side_only=bool(cdef and cdef.client_side and not cdef.actions), config=row.config_json or {}, ) ) return snapshots async def accept_crm_inbound_events( self, *, provider: str, verified: bool, payload: Any, provider_account_id: Optional[str] = None, claimed_project_id: Optional[UUID] = None, sink: Any = None, ): """Inbound CRM event plane. Not authorize_execution / crm.* invoke.""" from src.domain.connectors.crm_events import ( CRMAttributedConnection, ingest_verified_crm_inbound, ) from src.domain.connectors.crm_resolve import CRMConnectionSnapshot rows = await self._repo.list_active_connections_for_connector_key(provider) connections: list[CRMAttributedConnection] = [] for row, catalog in rows: if not row.project_id: continue connections.append( CRMAttributedConnection( project_id=row.project_id, workspace_id=row.workspace_id, connection=CRMConnectionSnapshot( connection_id=row.id, connector_key=catalog.key, status=row.status, has_secrets=bool(row.secrets_enc), permission_mode="allow", execution_domain="C", config=row.config_json or {}, ), ) ) return ingest_verified_crm_inbound( provider=provider, verified=verified, payload=payload, connections=connections, provider_account_id=provider_account_id, claimed_project_id=claimed_project_id, sink=sink, ) async def invoke_crm_contact( self, *, workspace_id: UUID, project_id: UUID, capability: str, payload: dict[str, Any], enabled_capabilities: Optional[set[str]] = None, provider_hint: Optional[str] = None, binding_connection_id: Optional[UUID] = None, approval_token: Optional[str] = None, enforce_invocable: bool = True, enqueue_execute: Optional[ Callable[..., Awaitable[dict[str, Any]]] ] = None, idempotency_key: Optional[str] = None, ): """Resolve crm.contact.*, re-authorize, enqueue via integration.execute.""" from src.domain.connectors.crm_contact import ( ACTION_BY_CAPABILITY, DispatchAuthorization, dispatch_crm_contact_plan, resolve_crm_contact, to_public_crm_contact_result, ) snapshots = await self._crm_snapshots( workspace_id=workspace_id, project_id=project_id, capability=capability, action_by_capability=ACTION_BY_CAPABILITY, ) resolution = resolve_crm_contact( capability=capability, payload=payload, connections=snapshots, enabled_capabilities=enabled_capabilities, provider_hint=provider_hint, binding_connection_id=binding_connection_id, approval_present=bool(approval_token), enforce_invocable=enforce_invocable, ) if not resolution.ok or resolution.plan is None: return to_public_crm_contact_result(resolution, capability=capability) decision = await self.authorize_execution( workspace_id=workspace_id, project_id=project_id, connection_id=resolution.plan.connection_id, action_slug=resolution.plan.action, enabled_capabilities=enabled_capabilities, approval_token=approval_token, ) dispatched = dispatch_crm_contact_plan( resolution.plan, DispatchAuthorization( allowed=decision.allowed, reason_code=decision.reason_code, message=decision.message, ), diagnostics=resolution.diagnostics, ) if not dispatched.ok or dispatched.plan is None: return to_public_crm_contact_result(dispatched, capability=capability) enqueue: dict[str, Any] | None = None if enqueue_execute is not None: plan = dispatched.plan enqueue = await enqueue_execute( connection_id=plan.connection_id, action=plan.action, input=dict(plan.resolved_input), provider_hint=plan.provider, idempotency_key=idempotency_key, approval_token=approval_token, ) _remember_outbound_crm_mutation(plan, capability, payload) return to_public_crm_contact_result( dispatched, capability=capability, enqueue=enqueue ) async def invoke_crm_company( self, *, workspace_id: UUID, project_id: UUID, capability: str, payload: dict[str, Any], enabled_capabilities: Optional[set[str]] = None, provider_hint: Optional[str] = None, binding_connection_id: Optional[UUID] = None, approval_token: Optional[str] = None, enforce_invocable: bool = True, enqueue_execute: Optional[ Callable[..., Awaitable[dict[str, Any]]] ] = None, idempotency_key: Optional[str] = None, ): """Resolve crm.company.*, re-authorize, enqueue via integration.execute.""" from src.domain.connectors.crm_company import ( ACTION_BY_CAPABILITY, dispatch_crm_company_plan, resolve_crm_company, to_public_crm_company_result, ) from src.domain.connectors.crm_resolve import DispatchAuthorization snapshots = await self._crm_snapshots( workspace_id=workspace_id, project_id=project_id, capability=capability, action_by_capability=ACTION_BY_CAPABILITY, ) resolution = resolve_crm_company( capability=capability, payload=payload, connections=snapshots, enabled_capabilities=enabled_capabilities, provider_hint=provider_hint, binding_connection_id=binding_connection_id, approval_present=bool(approval_token), enforce_invocable=enforce_invocable, ) if not resolution.ok or resolution.plan is None: return to_public_crm_company_result(resolution, capability=capability) decision = await self.authorize_execution( workspace_id=workspace_id, project_id=project_id, connection_id=resolution.plan.connection_id, action_slug=resolution.plan.action, enabled_capabilities=enabled_capabilities, approval_token=approval_token, ) dispatched = dispatch_crm_company_plan( resolution.plan, DispatchAuthorization( allowed=decision.allowed, reason_code=decision.reason_code, message=decision.message, ), diagnostics=resolution.diagnostics, ) if not dispatched.ok or dispatched.plan is None: return to_public_crm_company_result(dispatched, capability=capability) enqueue: dict[str, Any] | None = None if enqueue_execute is not None: plan = dispatched.plan enqueue = await enqueue_execute( connection_id=plan.connection_id, action=plan.action, input=dict(plan.resolved_input), provider_hint=plan.provider, idempotency_key=idempotency_key, approval_token=approval_token, ) _remember_outbound_crm_mutation(plan, capability, payload) return to_public_crm_company_result( dispatched, capability=capability, enqueue=enqueue ) async def invoke_crm_deal( self, *, workspace_id: UUID, project_id: UUID, capability: str, payload: dict[str, Any], enabled_capabilities: Optional[set[str]] = None, provider_hint: Optional[str] = None, binding_connection_id: Optional[UUID] = None, approval_token: Optional[str] = None, enforce_invocable: bool = True, enqueue_execute: Optional[ Callable[..., Awaitable[dict[str, Any]]] ] = None, idempotency_key: Optional[str] = None, ): """Resolve crm.deal.*, re-authorize, enqueue via integration.execute.""" from src.domain.connectors.crm_deal import ( ACTION_BY_CAPABILITY, dispatch_crm_deal_plan, resolve_crm_deal, to_public_crm_deal_result, ) from src.domain.connectors.crm_resolve import DispatchAuthorization snapshots = await self._crm_snapshots( workspace_id=workspace_id, project_id=project_id, capability=capability, action_by_capability=ACTION_BY_CAPABILITY, ) resolution = resolve_crm_deal( capability=capability, payload=payload, connections=snapshots, enabled_capabilities=enabled_capabilities, provider_hint=provider_hint, binding_connection_id=binding_connection_id, approval_present=bool(approval_token), enforce_invocable=enforce_invocable, ) if not resolution.ok or resolution.plan is None: return to_public_crm_deal_result(resolution, capability=capability) decision = await self.authorize_execution( workspace_id=workspace_id, project_id=project_id, connection_id=resolution.plan.connection_id, action_slug=resolution.plan.action, enabled_capabilities=enabled_capabilities, approval_token=approval_token, ) dispatched = dispatch_crm_deal_plan( resolution.plan, DispatchAuthorization( allowed=decision.allowed, reason_code=decision.reason_code, message=decision.message, ), diagnostics=resolution.diagnostics, ) if not dispatched.ok or dispatched.plan is None: return to_public_crm_deal_result(dispatched, capability=capability) enqueue: dict[str, Any] | None = None if enqueue_execute is not None: plan = dispatched.plan enqueue = await enqueue_execute( connection_id=plan.connection_id, action=plan.action, input=dict(plan.resolved_input), provider_hint=plan.provider, idempotency_key=idempotency_key, approval_token=approval_token, ) _remember_outbound_crm_mutation(plan, capability, payload) return to_public_crm_deal_result( dispatched, capability=capability, enqueue=enqueue ) async def invoke_crm_association( self, *, workspace_id: UUID, project_id: UUID, capability: str, payload: dict[str, Any], enabled_capabilities: Optional[set[str]] = None, provider_hint: Optional[str] = None, binding_connection_id: Optional[UUID] = None, approval_token: Optional[str] = None, enforce_invocable: bool = True, enqueue_execute: Optional[ Callable[..., Awaitable[dict[str, Any]]] ] = None, idempotency_key: Optional[str] = None, ): """Resolve crm.association.create, re-authorize, enqueue via integration.execute.""" from src.domain.connectors.crm_association import ( actions_for_kind, classify_relationship, dispatch_crm_association_plan, parse_association, resolve_crm_association, to_public_crm_association_result, ) from src.domain.connectors.crm_resolve import DispatchAuthorization parsed = parse_association(payload.get("association")) action_map = {"crm.association.create": {"hubspot": "hubspot.associate_objects"}} if not isinstance(parsed, str): kind = classify_relationship(parsed) if kind is not None: action_map = actions_for_kind(kind) snapshots = await self._crm_snapshots( workspace_id=workspace_id, project_id=project_id, capability=capability, action_by_capability=action_map, ) resolution = resolve_crm_association( capability=capability, payload=payload, connections=snapshots, enabled_capabilities=enabled_capabilities, provider_hint=provider_hint, binding_connection_id=binding_connection_id, approval_present=bool(approval_token), enforce_invocable=enforce_invocable, ) if not resolution.ok or resolution.plan is None: return to_public_crm_association_result(resolution, capability=capability) decision = await self.authorize_execution( workspace_id=workspace_id, project_id=project_id, connection_id=resolution.plan.connection_id, action_slug=resolution.plan.action, enabled_capabilities=enabled_capabilities, approval_token=approval_token, ) dispatched = dispatch_crm_association_plan( resolution.plan, DispatchAuthorization( allowed=decision.allowed, reason_code=decision.reason_code, message=decision.message, ), diagnostics=resolution.diagnostics, ) if not dispatched.ok or dispatched.plan is None: return to_public_crm_association_result(dispatched, capability=capability) enqueue: dict[str, Any] | None = None if enqueue_execute is not None: plan = dispatched.plan enqueue = await enqueue_execute( connection_id=plan.connection_id, action=plan.action, input=dict(plan.resolved_input), provider_hint=plan.provider, idempotency_key=idempotency_key, approval_token=approval_token, ) return to_public_crm_association_result( dispatched, capability=capability, enqueue=enqueue ) async def invoke_resource( self, *, workspace_id: UUID, project_id: UUID, capability: str, payload: dict, enabled_capabilities: set[str], enqueue_execute, provider_hint: str | None = None, binding_connection_id: UUID | None = None, approval_token: str | None = None, idempotency_key: str | None = None, ) -> dict: """Select customer resources, authorize, then use the existing enqueue seam.""" from src.domain.connectors.catalog import get_action, slugs_implementing from src.domain.connectors.resource_resolve import ( ResourceConnection, ResourceError, resolve_resource, validate_payload, ) def rejected(code): return {"capability": capability, "status": "rejected", "reason_code": code} try: validate_payload(capability, payload) snapshots = [] for row in await self._repo.list_connections_for_project(project_id): if row.workspace_id != workspace_id: continue catalog = await self._repo.get_connector_by_id(row.connector_id) if not catalog: continue permissions = {} for slug in slugs_implementing(capability): if get_action(slug)[0].key != catalog.key: continue permission = await self._repo.get_permission(row.id, slug) permissions[slug] = permission.mode if permission else "deny" snapshots.append(ResourceConnection( connection_id=row.id, connector_key=catalog.key, config=row.config_json or {}, status=row.status, has_secrets=bool(row.secrets_enc), scopes=set(row.scopes or []), permissions=permissions, )) plan = resolve_resource( capability, payload, snapshots, enabled_capabilities=enabled_capabilities, provider_hint=provider_hint, binding_connection_id=binding_connection_id, ) except ResourceError as exc: return rejected(exc.code) decision = await self.authorize_execution( workspace_id=workspace_id, project_id=project_id, connection_id=plan.connection_id, action_slug=plan.action, enabled_capabilities=enabled_capabilities, approval_token=approval_token, ) if not decision.allowed: return rejected(decision.reason_code or "connector_action_denied") result = await enqueue_execute( connection_id=plan.connection_id, action=plan.action, input=plan.input, provider_hint=plan.provider, approval_token=approval_token, idempotency_key=idempotency_key, ) if not result.get("ok"): return rejected(result.get("reason_code") or "provider_unavailable") # Enqueue acknowledges submission only. Never publish provider outputs or fake resource IDs. return {"capability": capability, "status": "queued"} async def authorize_execution( self, *, workspace_id: UUID, project_id: UUID, connection_id: UUID, action_slug: str, enabled_capabilities: Optional[set[str]] = None, approval_token: Optional[str] = None, ) -> ConnectorExecutionDecision: row = await self._repo.get_connection_for_project(project_id, connection_id) if not row or row.workspace_id != workspace_id: return ConnectorExecutionDecision( allowed=False, reason_code="connection_not_found", message="Connector connection not found for project", ) if row.status != "active": return ConnectorExecutionDecision( allowed=False, reason_code="connection_inactive", message="Connector connection is not active", connection=row, ) cdef = await self._connector_def_for_row(row) action_pair = get_action(action_slug) if not action_pair or action_pair[0].key != cdef.key: return ConnectorExecutionDecision( allowed=False, reason_code="unknown_connector_action", message="Connector action is not available for this connection", connection=row, connector=cdef, ) _connector, action = action_pair if getattr(row, "provider_connection_id", None): from src.domain.provider_auth.facades import execution_credentials try: credentials, scopes = await execution_credentials(self._repo._db, workspace_id=workspace_id, project_id=project_id, connection_id=row.provider_connection_id, connector_key=cdef.key) except ValidationError as exc: reason = str(exc) return ConnectorExecutionDecision(allowed=False, reason_code=reason if reason in {"scope_escalation_pending", "reauthentication_required"} else "authorization_required", message="Connect the required platform OAuth feature", connection=row, connector=cdef, action=action) row.scopes = scopes row.secrets_enc = {**(row.secrets_enc or {}), **{k: encrypt_value(v) for k, v in credentials.items()}} await self._push_connection_to_cloud(row, cdef, push_credentials=True) if cdef.secret_fields and not row.secrets_enc: return ConnectorExecutionDecision( allowed=False, reason_code="credentials_missing", message="Connector credentials are not configured", connection=row, connector=cdef, action=action, ) missing_scopes = sorted(set(action.required_scopes) - set(row.scopes or [])) if missing_scopes: return ConnectorExecutionDecision( allowed=False, reason_code="scope_missing", message=f"Connector connection is missing required scope(s): {', '.join(missing_scopes)}", connection=row, connector=cdef, action=action, ) missing_capabilities = sorted( set(action.required_capabilities) - (enabled_capabilities or set()) ) if missing_capabilities: return ConnectorExecutionDecision( allowed=False, reason_code="capability_not_enabled", message=( "Project is missing required connector capability/capabilities: " f"{', '.join(missing_capabilities)}" ), connection=row, connector=cdef, action=action, ) permission = await self._repo.get_permission(row.id, action_slug) mode = permission.mode if permission else "deny" if mode == "deny": return ConnectorExecutionDecision( allowed=False, reason_code="connector_action_denied", message="Connector action is denied by project policy", connection=row, connector=cdef, action=action, permission=permission, ) if mode == "ask": approval_claims = ( decode_connector_approval_token(approval_token) if approval_token else None ) if not approval_claims: return ConnectorExecutionDecision( allowed=False, reason_code="approval_required" if not approval_token else "invalid_approval_token", message=( "Connector action requires an approval token" if not approval_token else "Connector approval token is invalid or expired" ), connection=row, connector=cdef, action=action, permission=permission, ) expected_claims = { "ws": str(workspace_id), "pid": str(project_id), "cid": str(connection_id), "act": action_slug, } if any(approval_claims.get(key) != value for key, value in expected_claims.items()): return ConnectorExecutionDecision( allowed=False, reason_code="invalid_approval_token", message="Connector approval token scope does not match this execution request", connection=row, connector=cdef, action=action, permission=permission, ) return ConnectorExecutionDecision( allowed=True, connection=row, connector=cdef, action=action, permission=permission, ) async def upsert_from_project_integration( self, *, project_id: UUID, workspace_id: UUID, provider: str, status: str, config: Optional[dict], secrets_enc: Optional[dict], ) -> None: """Compatibility bridge from legacy project_integrations rows.""" cdef = get_connector(provider) if not cdef: return catalog_connector_id = await self._sync_connector_catalog_entry(cdef) row = await self._repo.get_connection_by_name(project_id, provider) fields = { "connector_id": catalog_connector_id, "workspace_id": workspace_id, "project_id": project_id, "name": provider, "status": "active" if status == "active" else "disconnected", "auth_data_ref": f"connector:{project_id}:{provider}" if secrets_enc else None, "scopes": [], "config_json": config or {}, "secrets_enc": secrets_enc, } if row: row = await self._repo.update_connection(row, **fields) else: row = await self._repo.create_connection(**fields) connection_id = row.id for action in cdef.actions: await self._repo.upsert_permission(connection_id, action.slug, mode="allow") await self._push_connection_to_cloud(row, cdef, push_credentials=bool(secrets_enc)) async def _require_project_connection( self, project_id: UUID, connection_id: UUID ) -> ConnectorConnection: row = await self._repo.get_connection_for_project(project_id, connection_id) if not row: raise NotFoundError("Connector connection not found") return row async def _connector_def_for_row(self, row: ConnectorConnection) -> ConnectorDef: connector = await self._repo.get_connector_by_id(row.connector_id) if connector: cdef = get_connector(connector.key) if cdef: return cdef for cdef in CONNECTORS.values(): persisted = await self._repo.get_connector_by_key(cdef.key) if persisted and persisted.id == row.connector_id: return cdef raise NotFoundError("Connector catalog entry not found") async def _connection_response(self, row: ConnectorConnection) -> ConnectorConnectionResponse: refetched = await self._repo.get_connection(row.id) if not refetched: raise NotFoundError("Connector connection not found") row = refetched cdef = await self._connector_def_for_row(row) permissions = await self._repo.list_permissions(row.id) return ConnectorConnectionResponse( provider_connection_id=row.provider_connection_id, id=row.id, connector_id=row.connector_id, connector_key=cdef.key, workspace_id=row.workspace_id, project_id=row.project_id, name=row.name, status=row.status, scopes=list(row.scopes or []), config=row.config_json, secrets_configured=sorted((row.secrets_enc or {}).keys()), permissions=[_permission_response(p) for p in permissions], created_at=row.created_at, updated_at=row.updated_at, ) async def _push_connection_to_cloud( self, row: ConnectorConnection, cdef: ConnectorDef, *, push_credentials: bool = False, ) -> None: # Snapshot ORM state before any external I/O — async session cannot lazy-load after await. connection_id = row.id workspace_id = row.workspace_id project_id = row.project_id name = row.name status = cloud_connector_runtime_status(row.status) auth_data_ref = row.auth_data_ref scopes = list(row.scopes or []) config = dict(row.config_json or {}) permissions = await self._repo.list_permissions(connection_id) await self._cloud.upsert_connector_connection( connection_id=str(connection_id), connector_key=cdef.key, workspace_id=str(workspace_id), project_id=str(project_id) if project_id else None, name=name, status=status, auth_data_ref=auth_data_ref, scopes=scopes, config=config, permissions=_cloud_permission_payloads( connection_id=connection_id, project_id=project_id, permissions=permissions, ), ) # Connector secrets remain in Core. Cloud requests them at execution # time through /v1/cloud/connector-credentials and does not persist a # second long-lived copy in its KV store. async def _validate_config_and_secrets( self, cdef: ConnectorDef, config: dict[str, Any], secrets: dict[str, str], *, require_all_secrets: bool = False, platform_grant: bool = False, ) -> None: unknown_config = set(config) - set(cdef.config_fields) if unknown_config: raise ConflictError( f"Unknown config field(s) for connector {cdef.key!r}: {', '.join(sorted(unknown_config))}" ) unknown_secrets = set(secrets) - set(cdef.secret_fields) if unknown_secrets: raise ConflictError( f"Unknown secret field(s) for connector {cdef.key!r}: {', '.join(sorted(unknown_secrets))}" ) _validate_required_connector_fields( cdef, config, secrets, require_all_secrets=require_all_secrets, platform_grant=platform_grant, ) # Supplied by the platform OAuth grant. Customers do not type these when Connect # already holds the client and refresh token. _PLATFORM_GRANT_CONFIG = frozenset({"client_id"}) _PLATFORM_GRANT_SECRETS = frozenset({"access_token", "refresh_token", "client_secret"}) def _decrypt_secret_map(secrets_enc: dict | None) -> dict[str, str]: return { key: decrypt_value(value) for key, value in (secrets_enc or {}).items() if value } def _validate_required_connector_fields( cdef: ConnectorDef, config: dict[str, Any], secrets: dict[str, str], *, require_all_secrets: bool = False, platform_grant: bool = False, ) -> None: """Enforce every field the connector's catalog entry declares as ``required=True``. Config fields are validated whenever a config value is being written at all — on create, or on an explicit config replace — since ``config_json`` is fully replaced, not merged. Required secret fields are only enforced on create/rotate with ``require_all_secrets=True``; a partial secret update just rejects blanking one out. """ missing_config = [ f.key for f in cdef.config_field_defs if f.required and not (platform_grant and f.key in _PLATFORM_GRANT_CONFIG) and not str((config or {}).get(f.key, "")).strip() ] for definition in cdef.config_field_defs: if (definition.input_type == "json" and definition.key in config and not isinstance(config[definition.key], dict)): raise ValidationError(f"{definition.label} must be a JSON object") if missing_config: raise ValidationError( f"{cdef.name} requires: {', '.join(missing_config)}" ) required_secret_keys = { f.key for f in cdef.secret_field_defs if f.required and not (platform_grant and f.key in _PLATFORM_GRANT_SECRETS) } if not required_secret_keys: return if require_all_secrets: missing_secrets = [ key for key in required_secret_keys if not str((secrets or {}).get(key, "")).strip() ] if missing_secrets: raise ValidationError( f"{cdef.name} requires: {', '.join(sorted(missing_secrets))}" ) return blanked = [ key for key in (secrets or {}) if key in required_secret_keys and not str(secrets.get(key, "")).strip() ] if blanked: raise ValidationError( f"{cdef.name} secret field(s) cannot be empty: {', '.join(blanked)}" ) def _permission_mode_for_create( cdef: ConnectorDef, permissions: dict[str, str], action_slug: str ) -> str: """Resolve permission mode on connection create (explicit map overrides defaults).""" if action_slug in permissions: return permissions[action_slug] # Preview / R2 test connectors: allow by default so dashboard flows work out of the box. if cdef.key == "cloudflare_r2" or cdef.status == "preview": return "allow" # Providing email credentials should start sending without a second permission step. if cdef.category == "email" and action_slug.endswith(".send_email"): return "allow" return "deny" def _require_connector(key: str) -> ConnectorDef: cdef = get_connector(key) if not cdef: raise NotFoundError(f"Unknown connector {key!r}") return cdef def _secret_ref(row: ConnectorConnection) -> str: return f"connector:{row.workspace_id}:{row.project_id}:{row.id}" def _connector_response(cdef: ConnectorDef) -> ConnectorResponse: from src.domain.provider_auth.manifests import authorization_manifest return ConnectorResponse( authorization=authorization_manifest(cdef.key), key=cdef.key, name=cdef.name, category=cdef.category, auth_type=cdef.auth_type, ownership_mode=cdef.ownership_mode, execution_domain=cdef.execution_domain, version=cdef.version, status=cdef.status, client_side=cdef.client_side, icon=cdef.icon, image_url=cdef.image_url, description=cdef.description, docs_url=cdef.docs_url, credential_check=credential_check_for(cdef.key).mode, credential_check_note=credential_check_for(cdef.key).note, config_fields=list(cdef.config_fields), secret_fields=list(cdef.secret_fields), config_field_defs=[ ConnectorFieldResponse( key=field.key, label=field.label, description=field.description, placeholder=field.placeholder, required=field.required, input_type=field.input_type, default=field.default, ) for field in cdef.config_field_defs ], secret_field_defs=[ ConnectorFieldResponse( key=field.key, label=field.label, description=field.description, placeholder=field.placeholder, required=field.required, input_type=field.input_type, default=field.default, ) for field in cdef.secret_field_defs ], changelog=[ ConnectorChangelogEntryResponse( version=e.version, date=e.date, notes=e.notes, ) for e in cdef.changelog ], actions=[ ConnectorActionResponse( action_key=a.key, slug=a.slug, name=a.name, description=a.description, input_schema=a.input_schema, output_schema=a.output_schema, execution_mode=a.execution_mode, required_scopes=list(a.required_scopes), required_capabilities=list(a.required_capabilities), implements=list(a.implements), ) for a in cdef.actions ], triggers=[ ConnectorTriggerResponse( trigger_key=t.key, event_type=t.event_type, name=t.name, webhook_required=t.webhook_required, polling_supported=t.polling_supported, schema=t.schema, ) for t in cdef.triggers ], ) def _permission_response(row: ConnectorPermission) -> ConnectorPermissionResponse: return ConnectorPermissionResponse( id=row.id, connection_id=row.connection_id, action_slug=row.action_slug, mode=row.mode, policy=row.policy_json, created_at=row.created_at, updated_at=row.updated_at, ) def _default_response(row: WorkspaceConnectorDefault) -> WorkspaceConnectorDefaultResponse: return WorkspaceConnectorDefaultResponse( id=row.id, workspace_id=row.workspace_id, connector_key=row.connector_key, config=row.config_json, secrets_configured=sorted((row.secrets_enc or {}).keys()), scopes=list(row.scopes or []), default_permissions=dict(row.default_permissions or {}), created_at=row.created_at, updated_at=row.updated_at, ) class ConnectorDefaultsService: """Manages workspace credential templates and copies them into project connections.""" def __init__( self, *, defaults_repo: WorkspaceConnectorDefaultRepository, connector_repo: ConnectorRepository, cloud_client: CloudControlClient, ) -> None: self._defaults = defaults_repo self._connectors = connector_repo self._cloud = cloud_client async def list_defaults( self, workspace_id: UUID ) -> list[WorkspaceConnectorDefaultResponse]: rows = await self._defaults.list_for_workspace(workspace_id) return [_default_response(r) for r in rows] async def get_default( self, workspace_id: UUID, connector_key: str ) -> WorkspaceConnectorDefaultResponse: _require_connector(connector_key) row = await self._defaults.get(workspace_id, connector_key) if not row: raise NotFoundError(f"No workspace default for connector {connector_key!r}") return _default_response(row) async def upsert_default( self, workspace_id: UUID, connector_key: str, body: UpsertWorkspaceConnectorDefaultRequest, ) -> WorkspaceConnectorDefaultResponse: cdef = _require_connector(connector_key) existing = await self._defaults.get(workspace_id, connector_key) merged_secrets = { key: value for key, value in { **_decrypt_secret_map(existing.secrets_enc if existing else None), **{key: value for key, value in body.secrets.items() if str(value).strip()}, }.items() if key in cdef.secret_fields and str(value).strip() } await _validate_config_and_secrets_static( cdef, body.config, merged_secrets, require_all_secrets=bool(merged_secrets) and not existing, ) await verify_connector_credentials(cdef.key, body.config, merged_secrets) valid_slugs = {a.slug for a in cdef.actions} invalid = set(body.default_permissions) - valid_slugs if invalid: raise ConflictError( f"Unknown action slug(s) for connector {connector_key!r}: {', '.join(sorted(invalid))}" ) secrets_enc = {k: encrypt_value(v) for k, v in merged_secrets.items()} row = await self._defaults.upsert( workspace_id, connector_key, config_json=body.config or None, secrets_enc=secrets_enc or None, scopes=list(body.scopes), default_permissions=dict(body.default_permissions), ) logger.info( "workspace_connector_default_upserted", workspace_id=str(workspace_id), connector=connector_key, ) return _default_response(row) async def delete_default(self, workspace_id: UUID, connector_key: str) -> None: row = await self._defaults.get(workspace_id, connector_key) if not row: raise NotFoundError(f"No workspace default for connector {connector_key!r}") await self._defaults.delete(row) logger.info( "workspace_connector_default_deleted", workspace_id=str(workspace_id), connector=connector_key, ) async def copy_to_project( self, *, workspace_id: UUID, project_id: UUID, connector_key: str, name: str, ) -> ConnectorConnectionResponse: """Copy workspace credential template into a new project-scoped connection. Validates the project belongs to the workspace before copying so credentials cannot be copied across workspace boundaries. """ from src.core.security.encryption import decrypt_value cdef = _require_connector(connector_key) default = await self._defaults.get(workspace_id, connector_key) if not default: raise NotFoundError( f"No workspace default configured for connector {connector_key!r}. " "Configure one under workspace connector defaults first." ) connector = await self._connectors.get_connector_by_key(connector_key) if not connector: raise NotFoundError(f"Connector {connector_key!r} is not available") catalog_connector_id = connector.id existing = await self._connectors.get_connection_by_name(project_id, name) if existing: raise ConflictError(f"Connector connection {name!r} already exists for this project") row = await self._connectors.create_connection( connector_id=catalog_connector_id, workspace_id=workspace_id, project_id=project_id, name=name, status="active", auth_data_ref=_secret_ref_for(workspace_id, project_id), scopes=list(default.scopes or []), config_json=default.config_json, secrets_enc=dict(default.secrets_enc) if default.secrets_enc else None, ) connection_id = row.id auth_data_ref = f"connector:{workspace_id}:{project_id}:{connection_id}" row = await self._connectors.update_connection(row, auth_data_ref=auth_data_ref) for action in cdef.actions: mode = (default.default_permissions or {}).get(action.slug, "deny") await self._connectors.upsert_permission(connection_id, action.slug, mode=mode) connector_svc = ConnectorService( repo=self._connectors, cloud_client=self._cloud ) await connector_svc._push_connection_to_cloud( row, cdef, push_credentials=bool(default.secrets_enc) ) logger.info( "connector_copied_from_workspace_default", project_id=str(project_id), workspace_id=str(workspace_id), connector=connector_key, ) row = await connector_svc._require_project_connection(project_id, connection_id) return await connector_svc._connection_response(row) def _secret_ref_for(workspace_id: UUID, project_id: UUID) -> str: return f"connector:{workspace_id}:{project_id}:copied" def _validate_config_and_secrets_static( cdef: ConnectorDef, config: dict, secrets: dict, *, require_all_secrets: bool = False, ) -> None: unknown_config = set(config) - set(cdef.config_fields) if unknown_config: raise ConflictError( f"Unknown config field(s) for connector {cdef.key!r}: {', '.join(sorted(unknown_config))}" ) unknown_secrets = set(secrets) - set(cdef.secret_fields) if unknown_secrets: raise ConflictError( f"Unknown secret field(s) for connector {cdef.key!r}: {', '.join(sorted(unknown_secrets))}" ) _validate_required_connector_fields( cdef, config, secrets, require_all_secrets=require_all_secrets )