diff --git a/BUNDLE_API.md b/BUNDLE_API.md index 24e40c07e00f..a33d70a88f74 100644 --- a/BUNDLE_API.md +++ b/BUNDLE_API.md @@ -224,6 +224,13 @@ the deserialize half is covered by ## Changelog +- **Headless integration context checks (2026-10-28).** `Component` checks the + selected capability's declared deployment contexts before resolving a + connection or running an integration action. Unsupported headless actions + raise `IntegrationContextUnsupportedError` with the existing + `action-unsupported` code instead of suggesting credential setup. Supported + contexts and public signatures are unchanged, so `BUNDLE_API_VERSION` remains `1`. + - **Decision model inputs (2026-10-22).** `ModelInput` accepts `model_type="decision"`: such a field never connects other models (`input_types=[]`, no external options), and `value="connect_other_models"` is refused. `FloatInput` gains `preserve_empty` diff --git a/design/provider-integrations/confluence.md b/design/provider-integrations/confluence.md new file mode 100644 index 000000000000..5c40421ac105 --- /dev/null +++ b/design/provider-integrations/confluence.md @@ -0,0 +1,26 @@ +# Confluence Cloud preview + +Status: draft +Owners (sign-off roles): platform owner, release owner + +The lfx-confluence package owns provider confluence and the separate confluence_integrations category. Legacy lfx-bundles Confluence components retain their category and class identities. + +| Capability | Auth and scope | Context | Delivery or API | +| --- | --- | --- | --- | +| Search pages by exact title or list one approved space | Delegated user and read:page:confluence | Hosted and self-managed | REST v2 pages with bounded explicit cursor | +| Read a current page and storage body | Delegated user and read:page:confluence | Hosted and self-managed | REST v2 page by ID | +| Create a current page in an approved space | Delegated user and read:page:confluence plus write:page:confluence | Hosted and self-managed | REST v2 page create with optional verified parent | +| Replace a page at the expected version | Delegated user and read:page:confluence plus write:page:confluence | Hosted and self-managed | REST v2 page update with optimistic content version | +| Observe newly created pages after a silent baseline | Delegated user, read:page:confluence and offline_access, background consent | Self-managed | Full bounded current-page snapshots | +| Observe higher versions of previously seen pages | Delegated user, read:page:confluence and offline_access, background consent | Self-managed | Full bounded current-page snapshots | + +All capabilities are preview. One operator-configured Cloud site and approved numeric spaces constrain targets. The account and verified grant bind credentials to that site. New page creation uses createdAt rather than first appearance in the space. Snapshot observations cannot recover intermediate edits, deletion history or all changes during access gaps. + +Data Center, Desktop, headless manual credentials, hosted triggers and push webhooks are unsupported. Setup, limits, rate handling, write ambiguity and recovery are documented in the [bundle README](../../src/bundles/confluence/README.md). Live tenant acceptance remains unverified. + +## Sign-off + +| Role | Name | Date | PR | +| --- | --- | --- | --- | +| platform owner | | | | +| release owner | | | | diff --git a/design/provider-integrations/contracts/confluence.v1.json b/design/provider-integrations/contracts/confluence.v1.json new file mode 100644 index 000000000000..1464b407d783 --- /dev/null +++ b/design/provider-integrations/contracts/confluence.v1.json @@ -0,0 +1,201 @@ +{ + "schema_version": 1, + "provider_id": "confluence", + "distribution": "lfx-confluence", + "manifest": { + "schema_version": 1, + "provider_id": "confluence", + "display_name": "Confluence Cloud", + "icon": "Confluence", + "docs_url": "https://github.com/langflow-ai/langflow/blob/d4dffe227d3ec238f430736d7ca6d16bbf084c9b/src/bundles/confluence/README.md#setup", + "auth_profiles": [ + { + "id": "user", + "kind": "oauth2_authorization_code", + "identity": "user_delegated", + "supports_pkce": false, + "supports_refresh": true, + "default_scopes": [ + "offline_access", + "read:confluence-user", + "read:page:confluence", + "write:page:confluence" + ], + "client_type_by_context": { + "hosted": "confidential", + "self_managed": "confidential" + }, + "owner_by_context": { + "hosted": "either", + "self_managed": "customer" + } + } + ], + "capabilities": [ + { + "id": "confluence.page.search", + "display_name": "Confluence: Search Pages", + "auth_profile_id": "user", + "identity": "user_delegated", + "required_scopes": [ + "read:page:confluence", + "read:confluence-user" + ], + "policy_keys": [ + "integrations.confluence.page.search" + ], + "substrate": "rest", + "maturity": "preview", + "deployment_contexts": [ + "hosted", + "self_managed" + ], + "risk": "read", + "component_ref": "ConfluenceSearchPagesComponent" + }, + { + "id": "confluence.page.read", + "display_name": "Confluence: Read Page", + "auth_profile_id": "user", + "identity": "user_delegated", + "required_scopes": [ + "read:page:confluence", + "read:confluence-user" + ], + "policy_keys": [ + "integrations.confluence.page.read" + ], + "substrate": "rest", + "maturity": "preview", + "deployment_contexts": [ + "hosted", + "self_managed" + ], + "risk": "read", + "component_ref": "ConfluenceReadPageComponent" + }, + { + "id": "confluence.page.create", + "display_name": "Confluence: Create Page", + "auth_profile_id": "user", + "identity": "user_delegated", + "required_scopes": [ + "read:page:confluence", + "write:page:confluence", + "read:confluence-user" + ], + "policy_keys": [ + "integrations.confluence.page.create" + ], + "substrate": "rest", + "maturity": "preview", + "deployment_contexts": [ + "hosted", + "self_managed" + ], + "risk": "write", + "component_ref": "ConfluenceCreatePageComponent" + }, + { + "id": "confluence.page.update", + "display_name": "Confluence: Update Page", + "auth_profile_id": "user", + "identity": "user_delegated", + "required_scopes": [ + "read:page:confluence", + "write:page:confluence", + "read:confluence-user" + ], + "policy_keys": [ + "integrations.confluence.page.update" + ], + "substrate": "rest", + "maturity": "preview", + "deployment_contexts": [ + "hosted", + "self_managed" + ], + "risk": "write", + "component_ref": "ConfluenceUpdatePageComponent" + }, + { + "id": "confluence.trigger.created", + "display_name": "Confluence: On Page Created", + "auth_profile_id": "user", + "identity": "user_delegated", + "required_scopes": [ + "read:page:confluence", + "offline_access", + "read:confluence-user" + ], + "policy_keys": [ + "integrations.confluence.trigger.created" + ], + "substrate": "rest", + "maturity": "preview", + "deployment_contexts": [ + "self_managed" + ], + "risk": "read", + "component_ref": "ConfluenceOnPageCreatedTriggerComponent" + }, + { + "id": "confluence.trigger.updated", + "display_name": "Confluence: On Page Updated", + "auth_profile_id": "user", + "identity": "user_delegated", + "required_scopes": [ + "read:page:confluence", + "offline_access", + "read:confluence-user" + ], + "policy_keys": [ + "integrations.confluence.trigger.updated" + ], + "substrate": "rest", + "maturity": "preview", + "deployment_contexts": [ + "self_managed" + ], + "risk": "read", + "component_ref": "ConfluenceOnPageUpdatedTriggerComponent" + } + ] + }, + "sources": [ + { + "url": "https://developer.atlassian.com/cloud/confluence/rest/v2/api-group-page/", + "title": "Confluence Cloud REST v2 page API", + "verified_on": "2026-10-22" + }, + { + "url": "https://developer.atlassian.com/cloud/confluence/oauth-2-3lo-apps/", + "title": "Confluence Cloud OAuth 2.0 3LO apps", + "verified_on": "2026-10-22" + }, + { + "url": "https://developer.atlassian.com/cloud/confluence/rest/v1/api-group-users/#api-wiki-rest-api-user-current-get", + "title": "Confluence current user API", + "verified_on": "2026-10-22" + } + ], + "transport_limitations": [ + "One operator-configured Confluence Cloud site and approved numeric spaces only.", + "Hosted actions and self-managed actions and polling are preview. Hosted triggers, Data Center, Desktop, headless manual credentials and webhooks are unsupported.", + "Polls observe latest versions after a silent baseline. They can miss intermediate edits, deletions, short-lived pages and access gaps.", + "Original createdAt separates newly created pages from old pages moved into a space. Newly visible existing pages are not update events until a later version is observed.", + "Each source retains at most 5000 distinct observed IDs and completes at most 50 requests of 100 pages. Partial snapshots never advance cursors." + ], + "evidence_paths": [ + "src/bundles/confluence/tests/test_contract.py", + "src/backend/tests/unit/services/triggers/test_confluence_provider.py", + "src/backend/tests/unit/services/triggers/test_confluence_lifecycle.py", + "src/bundles/confluence/scripts/verify_wheel.py", + "src/bundles/confluence/README.md", + "design/provider-integrations/confluence.md" + ], + "owner_acceptance": { + "status": "pending", + "note": "Platform and release owners still need to review this preview. Operator app registration, provider distribution requirements, real tenant authorization, live content writes and production delivery acceptance remain unverified." + } +} diff --git a/design/provider-integrations/index.json b/design/provider-integrations/index.json index c7eacc887d5c..68fdbee2e554 100644 --- a/design/provider-integrations/index.json +++ b/design/provider-integrations/index.json @@ -1,6 +1,10 @@ { "schema_version": 1, "providers": [ + { + "provider_id": "confluence", + "contract": "contracts/confluence.v1.json" + }, { "provider_id": "github", "contract": "contracts/github.v1.json" diff --git a/pyproject.toml b/pyproject.toml index 7034c8732ad1..f60c90cd028a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -45,6 +45,7 @@ dependencies = [ "lfx-google>=0.4.0,<1.0.0", "lfx-microsoft>=0.1.5,<1.0.0", "lfx-slack>=0.2.3,<1.0.0", + "lfx-confluence>=0.1.0,<1.0.0", "lfx-github>=0.1.0,<1.0.0", # langflow-extensions:bundle-deps-end ] @@ -134,6 +135,7 @@ lfx-serply = { workspace = true } lfx-serpingapi = { workspace = true } lfx-microsoft = { workspace = true } lfx-slack = { workspace = true } +lfx-confluence = { workspace = true } lfx-github = { workspace = true } # langflow-extensions:bundle-sources-end torch = { index = "pytorch-cpu" } @@ -175,6 +177,7 @@ members = [ "src/bundles/serpingapi", "src/bundles/microsoft", "src/bundles/slack", + "src/bundles/confluence", "src/bundles/github", # langflow-extensions:bundle-members-end ] diff --git a/scripts/ci/bundle_profile_locks/enterprise-hardened.lock.json b/scripts/ci/bundle_profile_locks/enterprise-hardened.lock.json index c39ba61d7a2d..3f5afca0484f 100644 --- a/scripts/ci/bundle_profile_locks/enterprise-hardened.lock.json +++ b/scripts/ci/bundle_profile_locks/enterprise-hardened.lock.json @@ -52,6 +52,16 @@ "requirement": "lfx-cohere==0.2.1", "resolved_version": "0.2.1" }, + { + "distribution": "lfx-confluence", + "extras": [], + "providers": [ + "lfx-confluence" + ], + "requested": ">=0.1.0,<1.0.0", + "requirement": "lfx-confluence==0.1.0", + "resolved_version": "0.1.0" + }, { "distribution": "lfx-datastax", "extras": [], @@ -190,12 +200,13 @@ "https://pypi.org/simple" ], "profile": "enterprise-hardened", - "profile_digest": "sha256:311190439f5c0ab6", + "profile_digest": "sha256:5c9c26a194741a34", "providers": [ "lfx-amazon", "lfx-anthropic", "lfx-azure", "lfx-cohere", + "lfx-confluence", "lfx-datastax", "lfx-docling", "lfx-github", diff --git a/scripts/ci/bundle_profiles.json b/scripts/ci/bundle_profiles.json index 766f13b4f847..de053ddf476a 100644 --- a/scripts/ci/bundle_profiles.json +++ b/scripts/ci/bundle_profiles.json @@ -163,6 +163,14 @@ "providers": [ "lfx-github" ] + }, + { + "distribution": "lfx-confluence", + "extras": [], + "version": ">=0.1.0,<1.0.0", + "providers": [ + "lfx-confluence" + ] } ] } diff --git a/scripts/ci/release_inventory_contract.json b/scripts/ci/release_inventory_contract.json index 391518777a5e..34a36a67de61 100644 --- a/scripts/ci/release_inventory_contract.json +++ b/scripts/ci/release_inventory_contract.json @@ -67,6 +67,7 @@ "lfx-anthropic", "lfx-azure", "lfx-cohere", + "lfx-confluence", "lfx-datastax", "lfx-docling", "lfx-github", @@ -104,6 +105,7 @@ "lfx-anthropic", "lfx-azure", "lfx-cohere", + "lfx-confluence", "lfx-datastax", "lfx-docling", "lfx-github", @@ -242,6 +244,7 @@ "lfx-arxiv", "lfx-azure", "lfx-cohere", + "lfx-confluence", "lfx-confluent", "lfx-datastax", "lfx-docling", diff --git a/src/backend/base/langflow/locales/en.json b/src/backend/base/langflow/locales/en.json index c8e0174be61d..5a8544cacf70 100644 --- a/src/backend/base/langflow/locales/en.json +++ b/src/backend/base/langflow/locales/en.json @@ -869,6 +869,62 @@ "components.conditionalrouter.inputs.true_case_message.info.f74733d4": "The message to pass if the condition is True.", "components.conditionalrouter.outputs.false_result.display_name.60a33e6c": "False", "components.conditionalrouter.outputs.true_result.display_name.3cbc87c7": "True", + "components.confluencecreatepage.description.5ef260e7": "Publish one page with a storage-format body in an approved space.", + "components.confluencecreatepage.display_name.79a74602": "Confluence: Create Page", + "components.confluencecreatepage.inputs.body_storage.display_name.e90fd905": "Storage Body", + "components.confluencecreatepage.inputs.connection.display_name.6c80e5d9": "Confluence Connection", + "components.confluencecreatepage.inputs.connection.info.6782d988": "Use an owned Confluence Cloud OAuth connection for the operator-configured site.", + "components.confluencecreatepage.inputs.parent_id.display_name.1191c8fa": "Parent Page ID", + "components.confluencecreatepage.inputs.space_id.display_name.e36720e0": "Space ID", + "components.confluencecreatepage.inputs.space_id.info.1ef18b7a": "An operator-approved numeric Confluence space ID.", + "components.confluencecreatepage.inputs.title.display_name.7e8cd205": "Title", + "components.confluencecreatepage.outputs.page.display_name.0a30a815": "Page", + "components.confluenceonpagecreatedtrigger.description.b92020fb": "Observe newly created pages after an initial approved-space baseline. Moved old pages do not count as creation.", + "components.confluenceonpagecreatedtrigger.display_name.84d6bf92": "Confluence: On Page Created", + "components.confluenceonpagecreatedtrigger.inputs.connection.display_name.6c80e5d9": "Confluence Connection", + "components.confluenceonpagecreatedtrigger.inputs.connection.info.6782d988": "Use an owned Confluence Cloud OAuth connection for the operator-configured site.", + "components.confluenceonpagecreatedtrigger.inputs.event_payload.display_name.4e1f49a9": "Event", + "components.confluenceonpagecreatedtrigger.inputs.event_payload.info.ed170c17": "Set by the server when this trigger fires: the JSON event the run was started with. Leave it empty; a manual run simply reports that no trigger event was present. Paste an event here to rehearse the downstream flow by hand.", + "components.confluenceonpagecreatedtrigger.inputs.space_id.display_name.e36720e0": "Space ID", + "components.confluenceonpagecreatedtrigger.inputs.space_id.info.13310ff4": "An approved numeric space ID with at most 5000 observed pages.", + "components.confluenceonpagecreatedtrigger.outputs.trigger_event.display_name.4e1f49a9": "Event", + "components.confluenceonpageupdatedtrigger.description.23e60e32": "Observe version changes of previously seen published pages. Intermediate edits can be missed.", + "components.confluenceonpageupdatedtrigger.display_name.810e9271": "Confluence: On Page Updated", + "components.confluenceonpageupdatedtrigger.inputs.connection.display_name.6c80e5d9": "Confluence Connection", + "components.confluenceonpageupdatedtrigger.inputs.connection.info.6782d988": "Use an owned Confluence Cloud OAuth connection for the operator-configured site.", + "components.confluenceonpageupdatedtrigger.inputs.event_payload.display_name.4e1f49a9": "Event", + "components.confluenceonpageupdatedtrigger.inputs.event_payload.info.ed170c17": "Set by the server when this trigger fires: the JSON event the run was started with. Leave it empty; a manual run simply reports that no trigger event was present. Paste an event here to rehearse the downstream flow by hand.", + "components.confluenceonpageupdatedtrigger.inputs.space_id.display_name.e36720e0": "Space ID", + "components.confluenceonpageupdatedtrigger.inputs.space_id.info.13310ff4": "An approved numeric space ID with at most 5000 observed pages.", + "components.confluenceonpageupdatedtrigger.outputs.trigger_event.display_name.4e1f49a9": "Event", + "components.confluencereadpage.description.48588649": "Read one current page and its storage body from an approved space.", + "components.confluencereadpage.display_name.3f266ca1": "Confluence: Read Page", + "components.confluencereadpage.inputs.connection.display_name.6c80e5d9": "Confluence Connection", + "components.confluencereadpage.inputs.connection.info.6782d988": "Use an owned Confluence Cloud OAuth connection for the operator-configured site.", + "components.confluencereadpage.inputs.page_id.display_name.4be1bb39": "Page ID", + "components.confluencereadpage.inputs.space_id.display_name.e36720e0": "Space ID", + "components.confluencereadpage.inputs.space_id.info.1ef18b7a": "An operator-approved numeric Confluence space ID.", + "components.confluencereadpage.outputs.page.display_name.0a30a815": "Page", + "components.confluencesearchpages.description.67961918": "List published pages in one approved space, optionally matching an exact title.", + "components.confluencesearchpages.display_name.b5ae9158": "Confluence: Search Pages", + "components.confluencesearchpages.inputs.connection.display_name.6c80e5d9": "Confluence Connection", + "components.confluencesearchpages.inputs.connection.info.6782d988": "Use an owned Confluence Cloud OAuth connection for the operator-configured site.", + "components.confluencesearchpages.inputs.cursor.display_name.705957be": "Next Page Cursor", + "components.confluencesearchpages.inputs.space_id.display_name.e36720e0": "Space ID", + "components.confluencesearchpages.inputs.space_id.info.1ef18b7a": "An operator-approved numeric Confluence space ID.", + "components.confluencesearchpages.inputs.title.display_name.2e7494ad": "Exact Title", + "components.confluencesearchpages.outputs.pages.display_name.9046da16": "Pages", + "components.confluenceupdatepage.description.25eafc9b": "Replace a published page only when its current version matches the expected version.", + "components.confluenceupdatepage.display_name.44eaa24e": "Confluence: Update Page", + "components.confluenceupdatepage.inputs.body_storage.display_name.e90fd905": "Storage Body", + "components.confluenceupdatepage.inputs.connection.display_name.6c80e5d9": "Confluence Connection", + "components.confluenceupdatepage.inputs.connection.info.6782d988": "Use an owned Confluence Cloud OAuth connection for the operator-configured site.", + "components.confluenceupdatepage.inputs.expected_version.display_name.2b99d04f": "Expected Current Version", + "components.confluenceupdatepage.inputs.page_id.display_name.4be1bb39": "Page ID", + "components.confluenceupdatepage.inputs.space_id.display_name.e36720e0": "Space ID", + "components.confluenceupdatepage.inputs.space_id.info.1ef18b7a": "An operator-approved numeric Confluence space ID.", + "components.confluenceupdatepage.inputs.title.display_name.7e8cd205": "Title", + "components.confluenceupdatepage.outputs.page.display_name.0a30a815": "Page", "components.conversationchain.description.cade5dbd": "Chain to have a conversation and load context from memory.", "components.conversationchain.display_name.9f9e9e71": "ConversationChain", "components.conversationchain.inputs.input_value.display_name.36ecb4f8": "Input", diff --git a/src/backend/base/langflow/services/triggers/listeners/adapters.py b/src/backend/base/langflow/services/triggers/listeners/adapters.py index ac183a45af62..7f45519b2d95 100644 --- a/src/backend/base/langflow/services/triggers/listeners/adapters.py +++ b/src/backend/base/langflow/services/triggers/listeners/adapters.py @@ -107,6 +107,10 @@ class ListenerContext: #: accept (Slack's ``hello``): that resets the failure count, and a socket #: dropped straight after it would then never back off. mark_connected: Callable[[], Awaitable[None]] | None = None + #: Polling health belongs to one source. Status updates are fenced against + #: its current settings and the listener lease. + source_failed: Callable[..., Awaitable[None]] | None = None + source_succeeded: Callable[..., Awaitable[None]] | None = None @runtime_checkable @@ -224,6 +228,7 @@ async def poll(self, ctx: ListenerContext) -> int: from langflow.services.database.models.trigger.model import Trigger from langflow.services.deps import session_scope from langflow.services.triggers.source_clients import source_lease + from langflow.services.triggers.source_errors import SourceConfigurationError from langflow.services.triggers.source_poll import collect_source, commit_source appended = 0 @@ -243,11 +248,29 @@ async def poll(self, ctx: ListenerContext) -> int: continue lease = await source_lease(session, row, family=FAMILY_TRIGGER_LISTENER) detached = Trigger(**row.model_dump()) - source_round = await collect_source(detached, lease) + try: + source_round = await collect_source(detached, lease) + except SourceConfigurationError as exc: + if ctx.source_failed is None: + raise + await ctx.source_failed( + trigger_id=snapshot.id, + message=str(exc), + expected_config=detached.config, + expected_kind=detached.kind, + ) + continue if ctx.stopping.is_set(): break async with session_scope() as session: appended += await commit_source(session, trigger_id=snapshot.id, source_round=source_round) + if ctx.source_succeeded is not None: + await ctx.source_succeeded( + trigger_id=snapshot.id, + expected_config=detached.config, + expected_kind=detached.kind, + observed_error=detached.last_error, + ) return appended diff --git a/src/backend/base/langflow/services/triggers/listeners/supervisor.py b/src/backend/base/langflow/services/triggers/listeners/supervisor.py index 13e5e41a9cd1..2b87ae6bb7b1 100644 --- a/src/backend/base/langflow/services/triggers/listeners/supervisor.py +++ b/src/backend/base/langflow/services/triggers/listeners/supervisor.py @@ -109,6 +109,7 @@ #: trigger back. These back off and retry like any other failure, and recover on #: their own the moment the process is configured correctly. _LOCAL_CONFIGURATION_REASONS = frozenset({"credential-undecryptable", "registration-unavailable"}) +_SOURCE_POLLING_ERROR_PREFIX = "Source polling failed. " def is_local_configuration_failure(exc: BaseException) -> bool: @@ -213,6 +214,8 @@ async def record_listener_error(session: AsyncSession, *, trigger_ids: list[UUID if not trigger_ids: return for row in (await session.exec(select(Trigger).where(col(Trigger.id).in_(trigger_ids)))).all(): + if (row.last_error or "").startswith(_SOURCE_POLLING_ERROR_PREFIX): + continue row.last_error = message row.updated_at = _now() session.add(row) @@ -225,6 +228,10 @@ async def clear_listener_error(session: AsyncSession, *, trigger_ids: list[UUID] return statement = select(Trigger).where(col(Trigger.id).in_(trigger_ids), col(Trigger.last_error).is_not(None)) for row in (await session.exec(statement)).all(): + if not (row.last_error or "").startswith( + ("The listener could not hold this connection", "The listener cannot read this connection") + ): + continue row.last_error = None row.updated_at = _now() session.add(row) @@ -857,6 +864,39 @@ async def mark_connected() -> None: if not worker.succeeded_since_failure: await self._succeeded(worker) + async def source_failed( + *, trigger_id: UUID, message: str, expected_config: dict[str, Any], expected_kind: str + ) -> None: + async with session_scope() as session: + row = await lock_target( + session, trigger_id=trigger_id, expected_config=expected_config, expected_kind=expected_kind + ) + row.last_error = f"{_SOURCE_POLLING_ERROR_PREFIX}{message}" + row.updated_at = _now() + session.add(row) + + async def source_succeeded( + *, trigger_id: UUID, expected_config: dict[str, Any], expected_kind: str, observed_error: str | None + ) -> None: + if observed_error is None and worker.succeeded_since_failure: + return + async with session_scope() as session: + row = await lock_target( + session, trigger_id=trigger_id, expected_config=expected_config, expected_kind=expected_kind + ) + if row.last_error == observed_error and (observed_error or "").startswith( + ( + _SOURCE_POLLING_ERROR_PREFIX, + "The listener could not hold this connection", + "The listener cannot read this connection", + ) + ): + row.last_error = None + row.updated_at = _now() + session.add(row) + if not worker.succeeded_since_failure: + await self._succeeded(worker) + return ListenerContext( connection_id=worker.connection_id, triggers=worker.triggers, @@ -865,6 +905,8 @@ async def mark_connected() -> None: resolve_credential=resolve_credential, stopping=worker.stopping, mark_connected=mark_connected, + source_failed=source_failed, + source_succeeded=source_succeeded, ) # ------------------------------------------------------------------ # diff --git a/src/backend/base/langflow/services/triggers/providers/confluence/__init__.py b/src/backend/base/langflow/services/triggers/providers/confluence/__init__.py new file mode 100644 index 000000000000..5f5a709ba946 --- /dev/null +++ b/src/backend/base/langflow/services/triggers/providers/confluence/__init__.py @@ -0,0 +1,29 @@ +"""Server runtime owned by the distinct Confluence integration bundle.""" + +from langflow.services.triggers.providers.confluence.collector import collect, normalize_config, prepare +from langflow.services.triggers.providers.confluence.oauth import OAUTH +from langflow.services.triggers.providers.registry import TriggerKind, TriggerProvider + +PROVIDER = TriggerProvider( + provider_id="confluence", + bundle="confluence_integrations", + kinds=tuple( + TriggerKind( + kind=kind, + component=component, + display_name=name, + config_fields=("connection", "space_id", "delivery_mode"), + required_scopes=("read:page:confluence", "offline_access"), + poll_mechanism="confluence.page.poll", + ) + for kind, component, name in ( + ("confluence.page_created", "ConfluenceOnPageCreatedTriggerComponent", "Confluence: On Page Created"), + ("confluence.page_updated", "ConfluenceOnPageUpdatedTriggerComponent", "Confluence: On Page Updated"), + ) + ), + normalize_config=normalize_config, + collect=collect, + prepare=prepare, + oauth=OAUTH, + api_origins=("https://api.atlassian.com",), +) diff --git a/src/backend/base/langflow/services/triggers/providers/confluence/collector.py b/src/backend/base/langflow/services/triggers/providers/confluence/collector.py new file mode 100644 index 000000000000..d2a87ef7170c --- /dev/null +++ b/src/backend/base/langflow/services/triggers/providers/confluence/collector.py @@ -0,0 +1,140 @@ +"""Observe complete bounded page snapshots and commit cursors through the shared runtime.""" + +from datetime import datetime, timezone + +from lfx.integrations.errors import IntegrationError +from lfx.integrations.models import ConnectionRef +from lfx_confluence.client import ConfluenceClient +from lfx_confluence.schema import ( + MAX_ITEMS, + MAX_PAGES, + MAX_VERSION, + approved_space, + cloud_id, + invalid, + list_params, + numeric_id, + page_list, + timestamp, +) + +from langflow.services.triggers.source_errors import SourceConfigurationError +from langflow.services.triggers.source_poll import SourceRound + +KINDS = {"confluence.page_created", "confluence.page_updated"} + + +def normalize_config(kind, config): + if ( + kind not in KINDS + or not isinstance(config, dict) + or config.keys() - {"connection", "space_id", "delivery_mode", "mechanism_id"} + ): + msg = "Unsupported Confluence trigger configuration." + raise ValueError(msg) + if (config.get("delivery_mode") or "poll") not in {"poll", "auto"}: + msg = "Confluence page sources support authenticated polling only." + raise ValueError(msg) + try: + ref = ConnectionRef.parse(config.get("connection", "")) + if ref.provider != "confluence": + raise ValueError + space = approved_space(config.get("space_id")) + except (ValueError, TypeError, IntegrationError): + msg = "Choose an owned Confluence connection and an approved space." + raise ValueError(msg) from None + return {**config, "connection": str(ref), "space_id": space, "delivery_mode": "poll"} + + +async def prepare(_session, trigger): + cloud_id() + approved_space((trigger.config or {}).get("space_id")) + if trigger.state != "active": + trigger.provider_state = {} + + +async def collect(trigger, lease): + config = normalize_config(trigger.kind, dict(trigger.config or {})) + previous = dict(trigger.provider_state or {}) + baseline = not previous.get("baseline_complete") + started = previous.get("started_at") or datetime.now(timezone.utc).isoformat() + threshold = timestamp(started) + seen = previous.get("pages", {}) + if not isinstance(seen, dict) or len(seen) > MAX_ITEMS: + msg = "The Confluence cursor is invalid. Re-enable the source for a fresh baseline." + raise invalid(msg) + for key, version in seen.items(): + numeric_id(key) + if type(version) is not int or not 1 <= version <= MAX_VERSION: + msg = "The Confluence source version cursor is invalid." + raise invalid(msg) + current, rows, cursors = {}, [], set() + cursor = "" + client = ConfluenceClient(lease) + try: + for _ in range(MAX_PAGES): + result = await client.request( + "GET", "/wiki/api/v2/pages", params=list_params(config["space_id"], cursor=cursor) + ) + pages, cursor = page_list(result, space=config["space_id"]) + for page in pages: + identifier = page["id"] + if identifier in current: + msg = "The Confluence space changed during pagination. Retry the complete snapshot." + raise invalid(msg) + current[identifier] = page["version"] + old = seen.get(identifier) + if old is not None and page["version"] < old: + msg = "Confluence returned an older page version. Retry the complete snapshot." + raise invalid(msg) + changed = ( + trigger.kind == "confluence.page_created" + and old is None + and timestamp(page["created_at"]) >= threshold + ) or (trigger.kind == "confluence.page_updated" and old is not None and page["version"] > old) + if changed and not baseline: + rows.append( + { + "provider": "confluence", + "resource": f"space:{config['space_id']}", + "id": identifier, + "version": "created" if trigger.kind == "confluence.page_created" else str(page["version"]), + "deleted": False, + "data": {**page, "observation": trigger.kind.rsplit(".", 1)[-1]}, + "session_key": f"confluence:{cloud_id()}:{identifier}", + } + ) + if len(current) > MAX_ITEMS: + msg = ( + "The Confluence space exceeds the 5000-page snapshot limit. " + "Choose a smaller approved space and re-enable this trigger." + ) + raise SourceConfigurationError(msg) + if not cursor: + break + if cursor in cursors: + msg = "Confluence repeated a pagination cursor." + raise invalid(msg) + cursors.add(cursor) + else: + msg = ( + "The Confluence space exceeds the 50-page request limit. " + "Choose a smaller approved space and re-enable this trigger." + ) + raise SourceConfigurationError(msg) + finally: + await client.aclose() + updated = {**seen, **current} + if len(updated) > MAX_ITEMS: + msg = "The source exceeded 5000 observed page IDs. Re-enable it for a fresh baseline." + raise SourceConfigurationError(msg) + return SourceRound( + items=rows, + cursor={"baseline_complete": True, "started_at": started, "pages": updated}, + previous_cursor=previous, + baseline=baseline, + snapshot_resource=None, + connection_id=trigger.connection_id, + mechanism_id=(trigger.config or {}).get("mechanism_id"), + config=dict(trigger.config or {}), + ) diff --git a/src/backend/base/langflow/services/triggers/providers/confluence/oauth.py b/src/backend/base/langflow/services/triggers/providers/confluence/oauth.py new file mode 100644 index 000000000000..9bfe3bf3ab35 --- /dev/null +++ b/src/backend/base/langflow/services/triggers/providers/confluence/oauth.py @@ -0,0 +1,117 @@ +"""Atlassian 3LO bound to the deployment's approved Confluence Cloud site.""" + +import re + +from lfx.integrations.errors import IntegrationError +from lfx_confluence.client import http_client, send +from lfx_confluence.schema import READ, SCOPES, USER, cloud_id + +from langflow.services.connection.oauth.adapter import OAuthAdapter +from langflow.services.connection.oauth.config import OAuthError + +MAX_TOKEN_LIFETIME = 86400 +MAX_ACCOUNT_LENGTH = 255 +MAX_TOKEN_LENGTH = 8192 + + +def endpoints(_registration): + return "https://auth.atlassian.com/authorize", "https://auth.atlassian.com/oauth/token" + + +def validate_registration(registration): + try: + site = cloud_id() + except IntegrationError: + msg = "The deployment Confluence Cloud site is not configured." + raise OAuthError(msg, reason="registration-unavailable") from None + if ( + registration.profile != "user" + or registration.client_type != "confidential" + or registration.private_key + or registration.context not in {"hosted", "self_managed"} + or registration.tenant != site + or set(registration.scopes) != SCOPES + or (registration.allowed_tenants and set(registration.allowed_tenants) != {site}) + ): + msg = "Confluence requires a confidential registration bound to the configured site and scopes." + raise OAuthError(msg, reason="registration-unavailable") + + +async def parse_token(registration, body, *, refresh): # noqa: ARG001 - rotating grants undergo the same verification + validate_registration(registration) + if ( + not isinstance(body, dict) + or not isinstance(body.get("access_token"), str) + or not body["access_token"] + or len(body["access_token"]) > MAX_TOKEN_LENGTH + or not re.fullmatch(r"[!-~]+", body["access_token"]) + or not isinstance(body.get("refresh_token"), str) + or not body["refresh_token"] + or len(body["refresh_token"]) > MAX_TOKEN_LENGTH + or type(body.get("expires_in")) is not int + or not 0 < body["expires_in"] <= MAX_TOKEN_LIFETIME + or not isinstance(body.get("scope"), str) + or not {READ, USER} <= set(body["scope"].split()) + ): + msg = "Atlassian did not return a valid refreshable Confluence grant." + raise OAuthError(msg, reason="provider-unavailable") + token = {key: body[key] for key in ("access_token", "refresh_token", "expires_in", "scope")} + try: + async with http_client() as http: + resources = await send(http, "GET", "/oauth/token/accessible-resources", body["access_token"]) + site_grant = ( + next( + ( + resource + for resource in resources + if ( + isinstance(resource, dict) + and resource.get("id") == cloud_id() + and isinstance(resource.get("scopes"), list) + and all(isinstance(scope, str) for scope in resource["scopes"]) + and {READ, USER} <= set(resource["scopes"]) + ) + ), + None, + ) + if isinstance(resources, list) + else None + ) + if site_grant is None: + msg = "The Confluence grant does not cover the configured Cloud site." + raise OAuthError(msg, reason="provider-unavailable") + token["scope"] = " ".join( + sorted(set(body["scope"].split()) & set(site_grant["scopes"]) | {"offline_access"}) + ) + account = await send( + http, "GET", f"/ex/confluence/{cloud_id()}/wiki/rest/api/user/current", body["access_token"] + ) + if ( + not isinstance(account, dict) + or not isinstance(account.get("accountId"), str) + or not 0 < len(account["accountId"]) <= MAX_ACCOUNT_LENGTH + or account.get("type") != "known" + ): + msg = "The Confluence grant has no authenticated account identity." + raise OAuthError(msg, reason="provider-unavailable") + except IntegrationError: + msg = "The Confluence grant could not be verified." + raise OAuthError(msg, reason="provider-unavailable") from None + identity = {"id": account["accountId"], "tenant_id": cloud_id()} + for field in ("displayName", "email"): + display = account.get(field) + if isinstance(display, str) and display.strip() and len(display) <= MAX_ACCOUNT_LENGTH: + identity["display"] = display.strip() + break + return token, identity + + +OAUTH = OAuthAdapter( + endpoints=endpoints, + allowed_origins=("https://auth.atlassian.com",), + validate_registration=validate_registration, + parse_token=parse_token, + authorization_params=lambda _registration: {"audience": "api.atlassian.com", "prompt": "consent"}, + token_encoding="json", # noqa: S106 - serialization format + supports_pkce=False, +) diff --git a/src/backend/tests/unit/services/triggers/confluence_testkit.py b/src/backend/tests/unit/services/triggers/confluence_testkit.py new file mode 100644 index 000000000000..ca91e27296a2 --- /dev/null +++ b/src/backend/tests/unit/services/triggers/confluence_testkit.py @@ -0,0 +1,74 @@ +"""Synthetic Confluence credentials and pages without server imports.""" + +from dataclasses import replace + +import pytest +from lfx.custom.custom_component.component import Component # noqa: F401 +from lfx.integrations.models import ( + ConnectionAccount, + ConnectionRef, + ConnectionResolutionRequest, + CredentialLease, + ResolvedCredential, +) +from lfx.services.authorization.base import ExecutionPrincipal +from pydantic import SecretStr + +SITE = "12345678-1234-1234-1234-123456789abc" +SCOPES = frozenset({"read:page:confluence", "write:page:confluence", "read:confluence-user", "offline_access"}) + + +def credential(**changes): + return replace( + ResolvedCredential( + access_token=SecretStr("synthetic-confluence-access"), + provider="confluence", + name="work", + granted_scopes=SCOPES, + scopes_verified=True, + identity="user_delegated", + account=ConnectionAccount(id="user-1", tenant_id=SITE), + ), + **changes, + ) + + +class Resolver: + def __init__(self, credentials): + self.credentials = credentials + self.calls = [] + + async def resolve(self, request): + self.calls.append(request) + return self.credentials[min(len(self.calls) - 1, len(self.credentials) - 1)] + + +def lease_for(*credentials): + resolver = Resolver(list(credentials) or [credential()]) + return CredentialLease( + resolver, + ConnectionResolutionRequest( + ref=ConnectionRef.parse("confluence/work"), + principal=ExecutionPrincipal(kind="actor", user_id="owner", interactive=True), + ), + ), resolver + + +def page(identifier="1", *, version=1, space="10", created="2026-10-21T10:00:00Z", title="Example"): + return { + "id": identifier, + "spaceId": space, + "title": title, + "status": "current", + "createdAt": created, + "version": {"number": version, "createdAt": "2026-10-22T10:00:00Z"}, + "body": {"storage": {"value": "
Example
"}}, + "_links": {"self": "https://evil.invalid/synthetic-confluence-access"}, + "unexpected_secret": "must-not-escape", # pragma: allowlist secret + } + + +@pytest.fixture(autouse=True) +def configured_site(monkeypatch): + monkeypatch.setenv("LANGFLOW_CONFLUENCE_CLOUD_ID", SITE) + monkeypatch.setenv("LANGFLOW_CONFLUENCE_SPACE_IDS", "10,20") diff --git a/src/backend/tests/unit/services/triggers/test_confluence_lifecycle.py b/src/backend/tests/unit/services/triggers/test_confluence_lifecycle.py new file mode 100644 index 000000000000..35e9fceec8bb --- /dev/null +++ b/src/backend/tests/unit/services/triggers/test_confluence_lifecycle.py @@ -0,0 +1,268 @@ +"""Installed Confluence adapter ownership, durable delivery and cleanup contracts.""" + +import asyncio +from dataclasses import replace +from pathlib import Path +from types import SimpleNamespace +from uuid import uuid4 + +import httpx +import pytest +from langflow.services.connection.oauth.broker import store_tokens +from langflow.services.database.models.connection.model import Connection +from langflow.services.database.models.trigger.model import Trigger, TriggerCleanup, TriggerEvent, TriggerSubscription +from langflow.services.database.models.user.model import User +from langflow.services.deps import session_scope +from langflow.services.triggers import dispatcher, source_runtime +from langflow.services.triggers.providers import registry +from langflow.services.triggers.reconciliation import reconcile_flow_triggers +from langflow.services.triggers.service import TriggerService +from langflow.services.triggers.source_cleanup import run_cleanup_pass +from langflow.services.triggers.source_poll import collect_source, commit_source +from lfx.extension import load_extension +from lfx.extension.bundle_registry import BundleRecord +from lfx.integrations.errors import IntegrationPolicyBlockedError, ProviderUnavailableError +from lfx.services.deps import get_integration_policy_service, get_policy_bundle_service +from lfx_confluence.client import ConfluenceClient +from sqlmodel import select + +from tests.unit.services.triggers.confluence_testkit import SITE, page + +pytestmark = pytest.mark.no_blockbuster + + +@pytest.fixture +def confluence(monkeypatch, client): # noqa: ARG001 - initialize application services + import lfx_confluence + from langflow.services.triggers.providers.confluence import PROVIDER, collector + from lfx.extension import bundle_registry + + monkeypatch.setenv("LANGFLOW_CONFLUENCE_CLOUD_ID", SITE) + monkeypatch.setenv("LANGFLOW_CONFLUENCE_SPACE_IDS", "10,20") + monkeypatch.setenv("LANGFLOW_CONNECTION_OAUTH_CONTEXT", "self_managed") + loaded = load_extension(Path(lfx_confluence.__file__).parent, distribution="lfx-confluence") + assert loaded.ok, loaded.errors + record = BundleRecord( + bundle="confluence_integrations", + extension_id="lfx-confluence", + extension_version="0.1.0", + slot="official", + components=tuple(loaded.components), + integrations=tuple(loaded.integrations), + ) + monkeypatch.setattr( + bundle_registry, + "get_default_registry", + lambda: SimpleNamespace( + snapshot=lambda: {"confluence_integrations": record}, list_integrations=lambda: loaded.integrations + ), + ) + monkeypatch.setattr(registry, "_cache_key", None) + monkeypatch.setattr(registry, "_providers", {}) + original = registry.metadata.entry_points + monkeypatch.setattr( + registry.metadata, + "entry_points", + lambda **kwargs: [ + SimpleNamespace( + name="confluence", dist=SimpleNamespace(metadata={"Name": "lfx-confluence"}), load=lambda: PROVIDER + ) + ] + if kwargs.get("group") == registry.ENTRY_POINT_GROUP + else original(**kwargs), + ) + state = SimpleNamespace(version=1, fail=False, calls=[]) + + def handler(request): + state.calls.append(request) + assert request.url.host == "api.atlassian.com" + assert request.url.path == f"/ex/confluence/{SITE}/wiki/api/v2/pages" + return ( + httpx.Response(503) if state.fail else httpx.Response(200, json={"results": [page(version=state.version)]}) + ) + + monkeypatch.setattr( + collector, "ConfluenceClient", lambda lease: ConfluenceClient(lease, transport=httpx.MockTransport(handler)) + ) + return state + + +async def connection(owner_id, *, allow=True, identity="user_delegated"): + async with session_scope() as session: + row = Connection( + provider_key="confluence", + name="work", + display_name="Confluence work", + owner_id=owner_id, + status="ready", + ownership_mode="user", + allow_non_interactive=allow, + executing_identity={"identity": identity, "account": {"id": "user-1", "tenant_id": SITE}}, + ) + session.add(row) + await session.flush() + await store_tokens( + session, + row, + {"version": 1, "access_token": "synthetic-confluence-access", "token_type": "Bearer"}, + ["read:page:confluence", "write:page:confluence", "read:confluence-user", "offline_access"], + ) + return row.id + + +async def reconcile(flow_id, owner_id): + data = { + "nodes": [ + { + "id": "confluence-source", + "data": { + "type": "ext:confluence_integrations:ConfluenceOnPageUpdatedTriggerComponent@official", + "node": { + "template": { + name: {"value": value} + for name, value in { + "connection": "confluence/work", + "space_id": "10", + }.items() + } + }, + }, + } + ], + "edges": [], + } + async with session_scope() as session: + await reconcile_flow_triggers(session, flow_id=flow_id, owner_id=owner_id, flow_data=data) + async with session_scope() as session: + return (await session.exec(select(Trigger).where(Trigger.flow_id == flow_id))).one() + + +async def activate(trigger_id): + async with session_scope() as session: + row = await session.get(Trigger, trigger_id) + await TriggerService().enable_and_activate(session, row=row) + assert row.state == "active" + assert row.config["mechanism_id"] == "confluence.page.poll" + + +async def test_registered_listener_adapter_baselines_delivers_dedupes_and_pauses( + confluence, trigger_owner, owned_flow, fake_background_service, monkeypatch +): + from langflow.services.triggers.listeners import adapters + from langflow.services.triggers.listeners.supervisor import ListenerSupervisor + + connection_id = await connection(trigger_owner) + row = await reconcile(owned_flow, trigger_owner) + assert row.last_error is None + await activate(row.id) + assert await source_runtime.sync_source(row.id) == 0 + confluence.version = 2 + monkeypatch.setattr(adapters, "_REGISTRY", {}) + adapters.register_builtin_adapters() + supervisor = ListenerSupervisor(holder="confluence-listener-contract") + + async def wait_for_delivery(): + while True: + async with session_scope() as session: + events = (await session.exec(select(TriggerEvent).where(TriggerEvent.trigger_id == row.id))).all() + if events: + assert len(events) == 1 + assert events[0].payload["data"]["id"] == "1" + assert "synthetic-confluence-access" not in repr(events[0].payload) + return + await asyncio.sleep(0.01) + + try: + await supervisor.reconcile() + assert isinstance(supervisor.workers[connection_id].adapter, adapters.ProviderSourcePollAdapter) + await asyncio.wait_for(wait_for_delivery(), timeout=10) + finally: + await supervisor.stop() + assert await source_runtime.sync_source(row.id) == 0 + assert await dispatcher.run_once(owner="confluence-dispatch") == 1 + assert len(fake_background_service.submits) == 1 + async with session_scope() as session: + current = await session.get(Trigger, row.id) + await TriggerService().disable(session, row=current) + assert not ( + await session.exec(select(TriggerSubscription).where(TriggerSubscription.trigger_id == row.id)) + ).all() + assert not (await session.exec(select(TriggerCleanup).where(TriggerCleanup.trigger_id == row.id))).all() + assert await run_cleanup_pass() == 0 + calls = len(confluence.calls) + assert await source_runtime.sync_source(row.id) == 0 + assert len(confluence.calls) == calls + + +async def test_failed_fetch_and_stale_replica_cannot_advance_cursor(confluence, trigger_owner, owned_flow): + from langflow.services.triggers.source_clients import source_lease + + await connection(trigger_owner) + row = await reconcile(owned_flow, trigger_owner) + await activate(row.id) + assert await source_runtime.sync_source(row.id) == 0 + async with session_scope() as session: + original = dict((await session.get(Trigger, row.id)).provider_state) + confluence.fail = True + with pytest.raises(ProviderUnavailableError): + await source_runtime.sync_source(row.id) + async with session_scope() as session: + assert (await session.get(Trigger, row.id)).provider_state == original + confluence.fail = False + confluence.version = 2 + async with session_scope() as session: + current = await session.get(Trigger, row.id) + lease = await source_lease(session, current, family="trigger_listener") + snapshot = Trigger(**current.model_dump()) + first = await collect_source(snapshot, lease) + second = await collect_source(snapshot, lease) + async with session_scope() as session: + assert await commit_source(session, trigger_id=row.id, source_round=first) == 1 + with pytest.raises(RuntimeError, match="cursor changed"): + async with session_scope() as session: + await commit_source(session, trigger_id=row.id, source_round=second) + + +@pytest.mark.parametrize("failure", ["owner", "consent", "identity", "scope", "policy"]) +async def test_activation_enforces_owner_consent_identity_scopes_policy(confluence, trigger_owner, owned_flow, failure): + owner = trigger_owner + if failure == "owner": + async with session_scope() as session: + other = User( + username=f"confluence-{uuid4().hex}", + password="not-a-login", # noqa: S106 # pragma: allowlist secret + is_active=True, + ) + session.add(other) + await session.flush() + owner = other.id + identifier = await connection( + owner, allow=failure != "consent", identity="bot" if failure == "identity" else "user_delegated" + ) + if failure == "scope": + async with session_scope() as session: + row = await session.get(Connection, identifier) + row.granted_scopes = [] + session.add(row) + row = await reconcile(owned_flow, trigger_owner) + bundle = get_policy_bundle_service() + original = bundle.snapshot + if failure == "policy": + bundle.publish( + replace( + original, + revision=original.revision + 1, + initialized=True, + blocked_integration_action_keys=frozenset({"integrations.confluence.trigger.updated"}), + ) + ) + get_integration_policy_service().invalidate() + try: + async with session_scope() as session: + current = await session.get(Trigger, row.id) + with pytest.raises((ValueError, IntegrationPolicyBlockedError)): + await TriggerService().enable_and_activate(session, row=current) + assert not confluence.calls + finally: + bundle.publish(replace(original, revision=bundle.snapshot.revision + 1)) + get_integration_policy_service().invalidate() diff --git a/src/backend/tests/unit/services/triggers/test_confluence_presentation.py b/src/backend/tests/unit/services/triggers/test_confluence_presentation.py new file mode 100644 index 000000000000..33523981b008 --- /dev/null +++ b/src/backend/tests/unit/services/triggers/test_confluence_presentation.py @@ -0,0 +1,9 @@ +"""Trigger settings use the same human labels as the component palette.""" + +from langflow.services.triggers.providers.confluence import PROVIDER +from lfx_confluence.components import confluence_integrations as components + + +def test_runtime_trigger_labels_match_the_palette(): + for kind in PROVIDER.kinds: + assert kind.display_name == getattr(components, kind.component).display_name diff --git a/src/backend/tests/unit/services/triggers/test_confluence_provider.py b/src/backend/tests/unit/services/triggers/test_confluence_provider.py new file mode 100644 index 000000000000..2b022d8e3dce --- /dev/null +++ b/src/backend/tests/unit/services/triggers/test_confluence_provider.py @@ -0,0 +1,431 @@ +"""Confluence snapshot, OAuth broker and invalid push contracts.""" + +from types import SimpleNamespace +from uuid import uuid4 + +import httpx +import pytest +from langflow.services.connection.oauth.config import OAuthError +from langflow.services.triggers.ingress.verifiers import IngressRejected, IngressRequest, IngressSecrets, verify +from langflow.services.triggers.providers.confluence import PROVIDER, collector, oauth +from langflow.services.triggers.source_errors import SourceConfigurationError +from lfx.integrations.errors import InvalidRequestError, ProviderUnavailableError +from lfx_confluence.client import ConfluenceClient +from lfx_confluence.schema import SCOPES + +from tests.unit.services.triggers.confluence_testkit import SITE, configured_site, lease_for, page # noqa: F401 + +pytestmark = pytest.mark.no_blockbuster + + +def trigger(kind="confluence.page_updated", state=None): + return SimpleNamespace( + kind=kind, + config={ + "connection": "confluence/work", + "space_id": "10", + "delivery_mode": "poll", + "mechanism_id": "confluence.page.poll", + }, + provider_state=state or {}, + connection_id=uuid4(), + ) + + +def patch_client(monkeypatch, handler): + monkeypatch.setattr( + collector, "ConfluenceClient", lambda lease: ConfluenceClient(lease, transport=httpx.MockTransport(handler)) + ) + + +async def test_baseline_is_silent_and_does_not_mutate_trigger(monkeypatch): + patch_client(monkeypatch, lambda _r: httpx.Response(200, json={"results": [page()]})) + node = trigger() + result = await collector.collect(node, lease_for()[0]) + assert result.baseline + assert not result.items + assert result.cursor["pages"] == {"1": 1} + assert node.provider_state == {} + + +@pytest.mark.parametrize( + ("kind", "observed", "expected"), + [ + ("confluence.page_updated", page(version=2), 1), + ("confluence.page_updated", page(), 0), + ("confluence.page_updated", page("2", version=5), 0), + ("confluence.page_created", page("2", created="2026-10-22T11:00:00Z"), 1), + ("confluence.page_created", page("2", created="2026-10-20T11:00:00Z"), 0), + ], +) +async def test_changes_deduplicate_and_old_moved_pages_are_not_new(monkeypatch, kind, observed, expected): + state = {"baseline_complete": True, "started_at": "2026-10-22T10:00:00Z", "pages": {"1": 1}} + node = trigger(kind, state) + patch_client(monkeypatch, lambda _r: httpx.Response(200, json={"results": [observed]})) + result = await collector.collect(node, lease_for()[0]) + assert len(result.items) == expected + assert node.provider_state == state + if expected: + assert result.items[0]["session_key"] == f"confluence:{SITE}:{observed['id']}" + node.provider_state = result.cursor + assert not (await collector.collect(node, lease_for()[0])).items + + +async def test_failed_later_page_cannot_advance_cursor(monkeypatch): + calls = [] + + def handler(request): + calls.append(request) + if len(calls) == 1: + return httpx.Response( + 200, json={"results": [page()], "_links": {"next": "/wiki/api/v2/pages?space-id=20&cursor=opaque"}} + ) + assert request.url.params["space-id"] == "10" + assert request.url.params["cursor"] == "opaque" + return httpx.Response(503) + + patch_client(monkeypatch, handler) + node = trigger() + with pytest.raises(ProviderUnavailableError): + await collector.collect(node, lease_for()[0]) + assert node.provider_state == {} + + +@pytest.mark.parametrize( + "payload", + [ + {"results": [page(space="20")]}, + {"results": [page(), page()]}, + {"results": [], "_links": {"next": "/wiki/api/v2/pages?cursor=repeat"}}, + {"results": [], "_links": {"next": "https://evil.invalid/wiki/api/v2/pages?cursor=x"}}, + ], +) +async def test_invalid_snapshots_do_not_advance(monkeypatch, payload): + patch_client(monkeypatch, lambda _r: httpx.Response(200, json=payload)) + node = trigger() + with pytest.raises(InvalidRequestError): + await collector.collect(node, lease_for()[0]) + assert node.provider_state == {} + + +async def test_version_regression_does_not_reset_durable_history(monkeypatch): + patch_client(monkeypatch, lambda _r: httpx.Response(200, json={"results": [page(version=2)]})) + node = trigger(state={"baseline_complete": True, "started_at": "2026-10-22T10:00:00Z", "pages": {"1": 3}}) + with pytest.raises(InvalidRequestError): + await collector.collect(node, lease_for()[0]) + assert node.provider_state["pages"]["1"] == 3 + + +async def test_snapshot_bound_and_cursor_retention(monkeypatch): + monkeypatch.setattr(collector, "MAX_PAGES", 2) + counter = 0 + + def handler(_request): + nonlocal counter + counter += 1 + return httpx.Response( + 200, json={"results": [page(str(counter))], "_links": {"next": f"/wiki/api/v2/pages?cursor={counter}"}} + ) + + patch_client(monkeypatch, handler) + node = trigger() + with pytest.raises(SourceConfigurationError, match="smaller approved space and re-enable"): + await collector.collect(node, lease_for()[0]) + assert node.provider_state == {} + + +@pytest.mark.parametrize("retained", [False, True]) +async def test_page_limits_report_a_source_configuration_error(monkeypatch, retained): + monkeypatch.setattr(collector, "MAX_ITEMS", 1) + state = {"baseline_complete": True, "started_at": "2026-10-22T10:00:00Z", "pages": {"1": 1}} + node = trigger(state=state if retained else {}) + pages = [page("2")] if retained else [page("1"), page("2")] + patch_client(monkeypatch, lambda _request: httpx.Response(200, json={"results": pages})) + with pytest.raises(SourceConfigurationError, match="5000"): + await collector.collect(node, lease_for()[0]) + assert node.provider_state == (state if retained else {}) + + +@pytest.mark.parametrize( + "config", + [ + {"connection": "other/work", "space_id": "10"}, + {"connection": "confluence/work", "space_id": "30"}, + {"connection": "confluence/work", "space_id": "10", "delivery_mode": "push"}, + {"connection": "confluence/work", "space_id": "10", "token": "no"}, + ], +) +def test_invalid_config_and_push_are_rejected(config): + with pytest.raises((ValueError, InvalidRequestError)): + collector.normalize_config("confluence.page_created", config) + + +def registration(): + return SimpleNamespace( + provider="confluence", + profile="user", + client_type="confidential", + private_key=None, + context="self_managed", + tenant=SITE, + allowed_tenants=[SITE], + scopes=list(SCOPES), + ) + + +def grant(): + return { + "access_token": "synthetic-confluence-access", + "refresh_token": "synthetic-rotating", + "expires_in": 3600, + "scope": " ".join(SCOPES - {"offline_access"}), + } + + +@pytest.mark.parametrize("refresh", [False, True]) +@pytest.mark.parametrize( + ("profile", "display"), + [ + ({"displayName": "Example Person", "email": "person@example.com"}, "Example Person"), + ({"email": "person@example.com"}, "person@example.com"), + ({"displayName": " ", "email": "person@example.com"}, "person@example.com"), + ({"displayName": "x" * 256}, None), + ({}, None), + ], +) +async def test_oauth_initial_and_rotating_grant_prove_site_and_account(monkeypatch, refresh, profile, display): + def handler(request): + if request.url.path == "/oauth/token/accessible-resources": + return httpx.Response(200, json=[{"id": SITE, "scopes": list(SCOPES)}]) + assert request.url.path == f"/ex/confluence/{SITE}/wiki/rest/api/user/current" + return httpx.Response(200, json={"accountId": "user-1", "type": "known", **profile}) + + monkeypatch.setattr(oauth, "http_client", lambda: httpx.AsyncClient(transport=httpx.MockTransport(handler))) + token, account = await oauth.parse_token(registration(), grant(), refresh=refresh) + assert account == {"id": "user-1", "tenant_id": SITE, **({"display": display} if display else {})} + assert "offline_access" in token["scope"] + + +@pytest.mark.parametrize("failure", ["site", "scopes", "account", "redirect"]) +async def test_oauth_wrong_tenant_scopes_or_anonymous_account_fail_closed(monkeypatch, failure): + def handler(request): + if failure == "redirect": + return httpx.Response(302, headers={"Location": "https://evil.invalid"}) + if request.url.path == "/oauth/token/accessible-resources": + return httpx.Response( + 200, + json=[ + { + "id": "other" if failure == "site" else SITE, + "scopes": [] if failure == "scopes" else list(SCOPES), + } + ], + ) + return httpx.Response(200, json={"accountId": "user-1", "type": "anonymous"}) + + monkeypatch.setattr(oauth, "http_client", lambda: httpx.AsyncClient(transport=httpx.MockTransport(handler))) + with pytest.raises(OAuthError): + await oauth.parse_token(registration(), grant(), refresh=True) + + +@pytest.mark.parametrize( + ("field", "value"), [("tenant", "other"), ("client_type", "public"), ("context", "desktop"), ("scopes", [])] +) +def test_registration_is_operator_owned_and_fixed_site(field, value): + selected = registration() + setattr(selected, field, value) + with pytest.raises(OAuthError): + oauth.validate_registration(selected) + + +def test_no_unsigned_webhook_is_accepted(monkeypatch): + monkeypatch.setattr("langflow.services.triggers.providers.registry.all_providers", lambda: (PROVIDER,)) + with pytest.raises(IngressRejected): + verify( + IngressRequest(provider="confluence", body=b'{"event":"page_created"}', headers={}, query={}), + IngressSecrets(), + tolerance_s=300, + ) + assert PROVIDER.verify is None + + +async def test_shared_oauth_broker_uses_json_and_preserves_rotation(monkeypatch): + from urllib.parse import parse_qs, urlsplit + + from langflow.services.connection.oauth import providers + from langflow.services.connection.oauth.config import OAuthRegistration + from langflow.services.triggers.providers import registry + + monkeypatch.setattr(registry, "all_providers", lambda: (PROVIDER,)) + selected = OAuthRegistration( + provider="confluence", + client_id="operator-app", + client_secret="synthetic-client-secret", # noqa: S106 # pragma: allowlist secret + redirect_uri="https://langflow.example/api/v1/connections/oauth/confluence/callback", + tenant=SITE, + allowed_tenants=[SITE], + scopes=list(SCOPES), + ) + query = parse_qs( + urlsplit( + providers.authorization_url(selected, state="opaque-state", verifier="verifier", scopes=list(SCOPES)) + ).query + ) + assert query["audience"] == ["api.atlassian.com"] + assert "code_challenge" not in query + calls = [] + + async def exchange(url, data, **kwargs): + calls.append(data) + assert url == "https://auth.atlassian.com/oauth/token" + assert kwargs == {"json_body": True} + assert "code_verifier" not in data + return grant() + + def handler(request): + if request.url.path == "/oauth/token/accessible-resources": + return httpx.Response(200, json=[{"id": SITE, "scopes": list(SCOPES)}]) + return httpx.Response(200, json={"accountId": "user-1", "type": "known"}) + + monkeypatch.setattr(providers, "_request", exchange) + monkeypatch.setattr(oauth, "http_client", lambda: httpx.AsyncClient(transport=httpx.MockTransport(handler))) + _, scopes, account = await providers.exchange( + selected, code="synthetic-code", verifier="verifier", previous_scopes=[] + ) + assert account["tenant_id"] == SITE + assert set(scopes) == SCOPES + await providers.exchange(selected, refresh_token="synthetic-previous", previous_scopes=list(SCOPES)) # noqa: S106 + assert calls[1]["grant_type"] == "refresh_token" + + +def test_saved_canvas_defaults_select_polling_and_invalid_space_is_a_validation_error(monkeypatch): + from langflow.services.triggers.providers import registry + from langflow.services.triggers.source_arming import normalize_source_config + + monkeypatch.setattr(registry, "all_providers", lambda: (PROVIDER,)) + raw = {"connection": "confluence/work", "space_id": "10", "delivery_mode": None} + assert normalize_source_config("confluence.page_updated", raw)["delivery_mode"] == "poll" + raw["space_id"] = "not-a-number" + with pytest.raises(ValueError, match="approved space"): + normalize_source_config("confluence.page_updated", raw) + + +@pytest.mark.parametrize( + "changes", + [ + {"access_token": "bad\nheader"}, + {"access_token": "a" * 8193}, + {"refresh_token": "r" * 8193}, + {"expires_in": True}, + {"scope": "read:page:confluence"}, + ], +) +async def test_malformed_oauth_tokens_fail_before_api_io(monkeypatch, changes): + monkeypatch.setattr(oauth, "http_client", lambda: pytest.fail("Unexpected API request")) + with pytest.raises(OAuthError): + await oauth.parse_token(registration(), {**grant(), **changes}, refresh=False) + + +@pytest.mark.parametrize("policy", ["all", "read", "write"]) +@pytest.mark.parametrize("refresh", [False, True]) +async def test_dialog_scope_defaults_authorize_and_exchange_without_overgrant(monkeypatch, policy, refresh): + import json + from pathlib import Path + from urllib.parse import parse_qs, urlsplit + + from langflow.services.connection.oauth import providers + from langflow.services.connection.oauth.config import OAuthRegistration + from langflow.services.triggers.providers import registry + from lfx.integrations.errors import ScopeMissingError + from lfx_confluence.client import ConfluenceClient + from lfx_confluence.schema import READ, USER, WRITE + + from tests.unit.services.triggers.confluence_testkit import credential, lease_for + + root = Path(__file__).resolve().parents[6] + scenarios = json.loads((root / "src/bundles/confluence/tests/fixtures/consent-scopes.json").read_text()) + selected_scopes = next(item["scopes"] for item in scenarios if item["policy"] == policy) + monkeypatch.setattr(registry, "all_providers", lambda: (PROVIDER,)) + selected = OAuthRegistration( + provider="confluence", + client_id="operator-app", + client_secret="synthetic-client-secret", # noqa: S106 # pragma: allowlist secret + redirect_uri="https://langflow.example/api/v1/connections/oauth/confluence/callback", + tenant=SITE, + allowed_tenants=[SITE], + scopes=list(SCOPES), + ) + query = parse_qs( + urlsplit( + providers.authorization_url( + selected, + state="opaque", + verifier="verifier", + scopes=selected_scopes, + ) + ).query + ) + assert set(query["scope"][0].split()) == set(selected_scopes) + + async def exchange(*_args, **_kwargs): + return { + "access_token": "synthetic-access", + "refresh_token": "synthetic-refresh", + "expires_in": 3600, + "scope": " ".join(selected_scopes), + } + + def handler(request): + if request.url.path == "/oauth/token/accessible-resources": + return httpx.Response(200, json=[{"id": SITE, "scopes": selected_scopes}]) + return httpx.Response(200, json={"accountId": "user-1", "type": "known"}) + + monkeypatch.setattr(providers, "_request", exchange) + monkeypatch.setattr(oauth, "http_client", lambda: httpx.AsyncClient(transport=httpx.MockTransport(handler))) + exchange_args = ( + {"refresh_token": "synthetic-previous"} if refresh else {"code": "synthetic-code", "verifier": "verifier"} + ) + _, granted, account = await providers.exchange(selected, previous_scopes=selected_scopes, **exchange_args) + assert set(granted) == set(selected_scopes) + assert account == {"id": "user-1", "tenant_id": SITE} + assert {READ, USER} <= set(granted) + if policy == "read": + calls = [] + client = ConfluenceClient( + lease_for(credential(granted_scopes=frozenset(granted)))[0], + transport=httpx.MockTransport(lambda request: calls.append(request)), + ) + try: + with pytest.raises(ScopeMissingError): + await client.request("PUT", "/wiki/api/v2/pages/1", body={}, scopes=frozenset({WRITE})) + finally: + await client.aclose() + assert calls == [] + + +async def test_site_grant_cannot_borrow_write_scope_from_another_site(monkeypatch): + from lfx_confluence.schema import READ, USER, WRITE + + def handler(request): + if request.url.path == "/oauth/token/accessible-resources": + return httpx.Response( + 200, + json=[ + {"id": SITE, "scopes": [READ, USER]}, + {"id": "other-site", "scopes": [READ, USER, WRITE]}, + ], + ) + return httpx.Response(200, json={"accountId": "user-1", "type": "known"}) + + monkeypatch.setattr(oauth, "http_client", lambda: httpx.AsyncClient(transport=httpx.MockTransport(handler))) + token, _ = await oauth.parse_token( + registration(), + { + "access_token": "synthetic-access", + "refresh_token": "synthetic-refresh", + "expires_in": 3600, + "scope": " ".join(SCOPES), + }, + refresh=False, + ) + assert set(token["scope"].split()) == {READ, USER, "offline_access"} diff --git a/src/backend/tests/unit/services/triggers/test_listener_source_isolation.py b/src/backend/tests/unit/services/triggers/test_listener_source_isolation.py new file mode 100644 index 000000000000..82e7240ae0cb --- /dev/null +++ b/src/backend/tests/unit/services/triggers/test_listener_source_isolation.py @@ -0,0 +1,209 @@ +"""Polling failures stay on the source whose configuration needs attention.""" + +from uuid import uuid4 + +import pytest +from langflow.services.database.models.connection.model import Connection +from langflow.services.database.models.trigger.model import Trigger, TriggerEvent +from langflow.services.deps import session_scope +from langflow.services.triggers.listeners import adapters +from langflow.services.triggers.listeners.supervisor import ListenerSupervisor +from langflow.services.triggers.source_errors import SourceConfigurationError +from langflow.services.triggers.source_poll import SourceRound +from lfx.integrations.errors import InvalidRequestError +from sqlmodel import select + +pytestmark = pytest.mark.no_blockbuster +KIND = "source_isolation_test" + + +@pytest.fixture +async def polling_sources(trigger_owner, make_trigger, monkeypatch): + async with session_scope() as session: + connection = Connection( + provider_key="selftest", + name=f"source_{uuid4().hex[:8]}", + display_name="Source isolation", + ownership_mode="user", + owner_id=trigger_owner, + status="ready", + allow_non_interactive=True, + ) + session.add(connection) + await session.flush() + connection_id = connection.id + ids = [ + await make_trigger(kind=KIND, provider="selftest", connection_id=connection_id, config={"source": value}) + for value in ("bad", "good") + ] + + async def lease(*_args, **_kwargs): + return None + + monkeypatch.setattr("langflow.services.triggers.source_clients.source_lease", lease) + monkeypatch.setattr(adapters.ProviderSourcePollAdapter, "_supports", staticmethod(lambda _snapshot: True)) + monkeypatch.setattr(adapters, "_REGISTRY", {}) + adapters.register_adapter( + kind=KIND, + mechanism=None, + factory=lambda _: adapters.ProviderSourcePollAdapter(interval_s=3600), + rebuild_on_config_change=False, + ) + supervisor = ListenerSupervisor(holder="source-isolation-test") + # Drive polls explicitly so state assertions do not race a timer. + monkeypatch.setattr(supervisor, "_spawn", lambda _worker: None) + await supervisor.reconcile() + worker = supervisor.workers[connection_id] + yield supervisor, worker, ids + await supervisor.stop() + + +def source_round(trigger): + return SourceRound( + items=[{"provider": "selftest", "resource": "items", "id": "1", "version": "1", "data": {}}], + cursor={"complete": True}, + previous_cursor=dict(trigger.provider_state), + baseline=False, + snapshot_resource=None, + connection_id=trigger.connection_id, + mechanism_id=None, + config=dict(trigger.config), + ) + + +async def test_configuration_error_isolated_from_healthy_source_and_clears_after_edit(polling_sources, monkeypatch): + supervisor, worker, (bad_id, good_id) = polling_sources + + async def collect(trigger, _lease): + if trigger.config["source"] == "bad": + msg = "The project exceeds 5000 issues. Add an issue type filter." + raise SourceConfigurationError(msg) + return source_round(trigger) + + monkeypatch.setattr("langflow.services.triggers.source_poll.collect_source", collect) + ctx = supervisor._context(worker) + assert await worker.adapter.poll(ctx) == 1 + # Both source orderings preserve the failing source's message. + worker.triggers.reverse() + assert await worker.adapter.poll(ctx) == 0 + async with session_scope() as session: + bad = await session.get(Trigger, bad_id) + good = await session.get(Trigger, good_id) + assert bad.last_error == "Source polling failed. The project exceeds 5000 issues. Add an issue type filter." + assert bad.state == good.state == "active" + assert bad.provider_state == {} + assert good.last_error is None + assert good.provider_state == {"complete": True} + assert len((await session.exec(select(TriggerEvent).where(TriggerEvent.trigger_id == good_id))).all()) == 1 + assert not (await session.exec(select(TriggerEvent).where(TriggerEvent.trigger_id == bad_id))).all() + bad.config = {"source": "fixed"} + session.add(bad) + await supervisor.reconcile() + assert await worker.adapter.poll(ctx) == 1 + async with session_scope() as session: + assert (await session.get(Trigger, bad_id)).last_error is None + assert (await session.get(Trigger, good_id)).last_error is None + assert worker.consecutive_failures == 0 + + +async def test_connection_recovery_keeps_another_sources_configuration_error(polling_sources, monkeypatch): + supervisor, worker, (bad_id, good_id) = polling_sources + + async def collect(trigger, _lease): + if trigger.id == bad_id: + msg = "Narrow this source's filter." + raise SourceConfigurationError(msg) + return source_round(trigger) + + monkeypatch.setattr("langflow.services.triggers.source_poll.collect_source", collect) + ctx = supervisor._context(worker) + assert await worker.adapter.poll(ctx) == 1 + worker.consecutive_failures = 6 + worker.succeeded_since_failure = False + async with session_scope() as session: + good = await session.get(Trigger, good_id) + good.last_error = "The listener could not hold this connection." + session.add(good) + assert await worker.adapter.poll(ctx) == 0 + async with session_scope() as session: + assert (await session.get(Trigger, bad_id)).last_error == "Source polling failed. Narrow this source's filter." + assert (await session.get(Trigger, good_id)).last_error is None + assert worker.consecutive_failures == 0 + + +async def test_provider_input_error_is_not_copied_into_source_status(polling_sources, monkeypatch): + supervisor, worker, ids = polling_sources + + async def collect(_trigger, _lease): + msg = "An untrusted provider message with private details" + raise InvalidRequestError(msg) + + monkeypatch.setattr("langflow.services.triggers.source_poll.collect_source", collect) + with pytest.raises(InvalidRequestError): + await worker.adapter.poll(supervisor._context(worker)) + async with session_scope() as session: + for identifier in ids: + assert (await session.get(Trigger, identifier)).last_error is None + + +@pytest.mark.parametrize( + ("banner", "expected"), + [ + ("Source polling failed. Old filter failed.", None), + ("The listener could not hold this connection after 6 attempts.", None), + ("Source reconciliation failed. Webhook setup failed.", "Source reconciliation failed. Webhook setup failed."), + ], +) +async def test_success_clears_own_persisted_errors_without_erasing_other_workers( + polling_sources, monkeypatch, banner, expected +): + supervisor, worker, (_bad_id, good_id) = polling_sources + worker.triggers[:] = [snapshot for snapshot in worker.triggers if snapshot.id == good_id] + async with session_scope() as session: + row = await session.get(Trigger, good_id) + row.last_error = banner + session.add(row) + + async def collect(trigger, _lease): + return source_round(trigger) + + monkeypatch.setattr("langflow.services.triggers.source_poll.collect_source", collect) + assert await worker.adapter.poll(supervisor._context(worker)) == 1 + async with session_scope() as session: + assert (await session.get(Trigger, good_id)).last_error == expected + + +async def test_stale_source_error_cannot_overwrite_changed_configuration(polling_sources, monkeypatch): + supervisor, worker, (bad_id, _good_id) = polling_sources + worker.triggers[:] = [snapshot for snapshot in worker.triggers if snapshot.id == bad_id] + + async def collect(trigger, _lease): + async with session_scope() as session: + row = await session.get(Trigger, trigger.id) + row.config = {"source": "fixed"} + session.add(row) + msg = "Old settings failed." + raise SourceConfigurationError(msg) + + monkeypatch.setattr("langflow.services.triggers.source_poll.collect_source", collect) + with pytest.raises(adapters.ListenerDeliveryRejectedError): + await worker.adapter.poll(supervisor._context(worker)) + async with session_scope() as session: + assert (await session.get(Trigger, bad_id)).last_error is None + + +async def test_success_does_not_clear_a_concurrent_source_error(polling_sources, monkeypatch): + supervisor, worker, (_bad_id, good_id) = polling_sources + worker.triggers[:] = [snapshot for snapshot in worker.triggers if snapshot.id == good_id] + + async def collect(trigger, _lease): + async with session_scope() as session: + row = await session.get(Trigger, trigger.id) + row.last_error = "Source polling failed. A newer poll failed." + session.add(row) + return source_round(trigger) + + monkeypatch.setattr("langflow.services.triggers.source_poll.collect_source", collect) + assert await worker.adapter.poll(supervisor._context(worker)) == 1 + async with session_scope() as session: + assert (await session.get(Trigger, good_id)).last_error == "Source polling failed. A newer poll failed." diff --git a/src/backend/tests/unit/test_lfx_bundles_extras.py b/src/backend/tests/unit/test_lfx_bundles_extras.py index 87ea2acb1bb0..8a1a9a4ad7b9 100644 --- a/src/backend/tests/unit/test_lfx_bundles_extras.py +++ b/src/backend/tests/unit/test_lfx_bundles_extras.py @@ -14,16 +14,18 @@ 3. ``all-no-torch`` is exactly ``all`` minus the torch-pulling providers (``TORCH_EXTRAS``), giving a torch-free full-provider install, 4. normalized extra keys are collision-free, - 5. the metapackage provider set stays disjoint from the graduated - partner distributions (no double-ship; manifest would shadow), + 5. metapackage bundle categories stay disjoint from categories declared + by partner manifests, preventing one distribution from shadowing another, 6. retired ALTK extras cannot reinstall the removed toolkit dependency. """ from __future__ import annotations +import json import re from pathlib import Path +import pytest from packaging.requirements import Requirement try: @@ -118,20 +120,73 @@ def test_normalized_extra_keys_are_collision_free() -> None: assert not dupes, f"provider names collide after PEP 685 normalization: {sorted(dupes)}" +def _partner_bundle_names(bundles_dir: Path) -> set[str]: + """Read registry categories from shipped manifests rather than workspace labels.""" + names = set() + for child in sorted(bundles_dir.iterdir()): + project_path = child / "pyproject.toml" + if not child.is_dir() or child.name == "lfx-bundles" or not project_path.is_file(): + continue + with project_path.open("rb") as file: + project = tomllib.load(file) + manifests = [ + json.loads(path.read_text(encoding="utf-8")) for path in sorted(child.glob("src/*/extension.json")) + ] + embedded = project.get("tool", {}).get("langflow", {}).get("extension") + if embedded is not None: + manifests.append(embedded) + assert manifests, f"partner distribution {child.name} has no extension manifest" + for manifest in manifests: + names.update(bundle["name"] for bundle in manifest.get("bundles", [])) + return names + + def test_metapackage_providers_disjoint_from_graduated_partners() -> None: - """A provider must ship from exactly one distribution. + """A registry category must ship from exactly one distribution. - Graduated ``lfx-Example
"}}, + "_links": {"self": "https://evil.invalid/synthetic-confluence-access"}, + "unexpected_secret": "must-not-escape", # pragma: allowlist secret + } + + +@pytest.fixture(autouse=True) +def configured_site(monkeypatch): + monkeypatch.setenv("LANGFLOW_CONFLUENCE_CLOUD_ID", SITE) + monkeypatch.setenv("LANGFLOW_CONFLUENCE_SPACE_IDS", "10,20") diff --git a/src/bundles/confluence/tests/fixtures/consent-scopes.json b/src/bundles/confluence/tests/fixtures/consent-scopes.json new file mode 100644 index 000000000000..472cb57ecebb --- /dev/null +++ b/src/bundles/confluence/tests/fixtures/consent-scopes.json @@ -0,0 +1,28 @@ +[ + { + "policy": "all", + "scopes": [ + "offline_access", + "read:confluence-user", + "read:page:confluence", + "write:page:confluence" + ] + }, + { + "policy": "read", + "scopes": [ + "offline_access", + "read:confluence-user", + "read:page:confluence" + ] + }, + { + "policy": "write", + "scopes": [ + "offline_access", + "read:confluence-user", + "read:page:confluence", + "write:page:confluence" + ] + } +] diff --git a/src/bundles/confluence/tests/test_client_transport.py b/src/bundles/confluence/tests/test_client_transport.py new file mode 100644 index 000000000000..4dfc16e4a546 --- /dev/null +++ b/src/bundles/confluence/tests/test_client_transport.py @@ -0,0 +1,31 @@ +"""Default transport construction and safe connection failures.""" + +import pytest +from lfx.integrations.errors import IntegrationNetworkError +from lfx.utils import ssrf_httpx +from lfx.utils.ssrf_protection import SSRFProtectionError +from lfx.utils.ssrf_transport import SSRFProtectedTransport +from lfx_confluence.client import http_client + + +@pytest.mark.parametrize("protection_enabled", [True, False]) +async def test_default_transport_keeps_redirects_disabled(monkeypatch, protection_enabled): + monkeypatch.setattr(ssrf_httpx, "is_ssrf_protection_enabled", lambda: protection_enabled) + monkeypatch.setattr(ssrf_httpx, "_validate_and_resolve_strict_url", lambda url: (url, ["93.184.216.34"])) + async with http_client() as client: + assert client.follow_redirects is False + assert client.trust_env is False + if protection_enabled: + assert isinstance(client._transport, SSRFProtectedTransport) + assert client._transport.pinned_ips == {"api.atlassian.com": ["93.184.216.34"]} + + +async def test_default_transport_dns_failure_is_normalized(monkeypatch): + def unavailable(_url): + msg = "private DNS diagnostic" + raise SSRFProtectionError(msg) + + monkeypatch.setattr(ssrf_httpx, "_validate_and_resolve_strict_url", unavailable) + with pytest.raises(IntegrationNetworkError, match=r"could not reach api\.atlassian\.com") as error: + http_client() + assert "private DNS diagnostic" not in str(error.value) diff --git a/src/bundles/confluence/tests/test_contract.py b/src/bundles/confluence/tests/test_contract.py new file mode 100644 index 000000000000..65f033696258 --- /dev/null +++ b/src/bundles/confluence/tests/test_contract.py @@ -0,0 +1,418 @@ +"""Standalone installed manifest, action and transport contracts.""" + +import importlib.abc +import json +import sys +from pathlib import Path + +import httpx +import pytest +from conftest import SITE, credential, lease_for, page +from lfx.extension import load_extension, validate_extension +from lfx.integrations.errors import ( + AuthExpiredError, + ConnectionNotAuthorizedError, + IntegrationNetworkError, + InvalidRequestError, + ProviderUnavailableError, + RateLimitedError, + ResourceNotFoundError, + ScopeMissingError, +) +from lfx.integrations.models import ConnectionAccount +from lfx_confluence.base import ConfluenceActionComponent +from lfx_confluence.client import ConfluenceClient +from lfx_confluence.components.confluence_integrations import ( + ConfluenceCreatePageComponent, + ConfluenceOnPageCreatedTriggerComponent, + ConfluenceOnPageUpdatedTriggerComponent, + ConfluenceReadPageComponent, + ConfluenceSearchPagesComponent, + ConfluenceUpdatePageComponent, +) +from lfx_confluence.manifest import load_manifest +from lfx_confluence.schema import approved_space, cloud_id, next_cursor, numeric_id + + +@pytest.mark.parametrize( + "component", + [ + ConfluenceSearchPagesComponent, + ConfluenceReadPageComponent, + ConfluenceCreatePageComponent, + ConfluenceUpdatePageComponent, + ConfluenceOnPageCreatedTriggerComponent, + ConfluenceOnPageUpdatedTriggerComponent, + ], +) +def test_palette_and_connection_dialog_use_the_same_provider_labels(component): + manifest = load_manifest() + capability = next(item for item in manifest.capabilities if item.component_ref == component.__name__) + assert capability.display_name == component.display_name + assert component.display_name.startswith("Confluence: ") + assert component.icon == manifest.icon == "Confluence" + + +async def run(monkeypatch, cls, method, handler, **inputs): + lease, _ = lease_for() + requests = [] + + def record(request): + requests.append(request) + assert request.url.host == "api.atlassian.com" + assert request.url.path.startswith(f"/ex/confluence/{SITE}/wiki/api/v2/pages") + return handler(request) + + monkeypatch.setattr( + ConfluenceActionComponent, + "client", + lambda _self: ConfluenceClient(lease, transport=httpx.MockTransport(record)), + ) + return (await getattr(cls(connection="confluence/work", **inputs), method)()).data, requests + + +async def test_read_page_retains_only_explicit_fields(monkeypatch): + def handler(request): + assert request.url.params["body-format"] == "storage" + return httpx.Response(200, json=page()) + + data, calls = await run(monkeypatch, ConfluenceReadPageComponent, "read_page", handler, space_id="10", page_id="1") + assert data["body_storage"] == "Example
" + assert "unexpected_secret" not in str(data) + assert "synthetic-confluence-access" not in str(data) + assert len(calls) == 1 + + +async def test_search_exact_title_in_approved_space_with_cursor(monkeypatch): + title = 'Exact " title' + + def handler(request): + assert dict(request.url.params) == { + "space-id": "10", + "status": "current", + "sort": "id", + "subtype": "page", + "limit": "100", + "title": title, + "cursor": "opaque", + } + return httpx.Response( + 200, json={"results": [page(title=title)], "_links": {"next": "/wiki/api/v2/pages?cursor=next"}} + ) + + data, _ = await run( + monkeypatch, + ConfluenceSearchPagesComponent, + "search_pages", + handler, + space_id="10", + title=title, + cursor="opaque", + ) + assert data["next_cursor"] == "next" + + +async def test_create_checks_parent_space_and_storage_payload(monkeypatch): + def handler(request): + if request.method == "GET": + assert request.url.path.endswith("/1") + return httpx.Response(200, json=page()) + assert json.loads(request.content) == { + "spaceId": "10", + "status": "current", + "title": "New", + "parentId": "1", + "body": {"representation": "storage", "value": "New
"}, + } + return httpx.Response(201, json=page("2", title="New")) + + data, calls = await run( + monkeypatch, + ConfluenceCreatePageComponent, + "create_page", + handler, + space_id="10", + title="New", + parent_id="1", + body_storage="New
", + ) + assert data["id"] == "2" + assert len(calls) == 2 + + +async def test_update_requires_expected_version_and_increments_once(monkeypatch): + def handler(request): + if request.method == "GET": + return httpx.Response(200, json=page(version=4)) + body = json.loads(request.content) + assert body == { + "id": "1", + "status": "current", + "title": "Changed", + "body": {"representation": "storage", "value": ""}, + "version": {"number": 5}, + } + return httpx.Response(200, json=page(version=5)) + + data, calls = await run( + monkeypatch, + ConfluenceUpdatePageComponent, + "update_page", + handler, + space_id="10", + page_id="1", + expected_version=4, + title="Changed", + body_storage="", + ) + assert data["version"] == 5 + assert len(calls) == 2 + + +@pytest.mark.parametrize("failure", ["version", "space", "target"]) +async def test_update_mismatch_sends_no_write(monkeypatch, failure): + def handler(request): + assert request.method == "GET" + return httpx.Response( + 200, + json=page( + "2" if failure == "target" else "1", + version=3 if failure == "version" else 4, + space="20" if failure == "space" else "10", + ), + ) + + with pytest.raises(InvalidRequestError): + await run( + monkeypatch, + ConfluenceUpdatePageComponent, + "update_page", + handler, + space_id="10", + page_id="1", + expected_version=4, + title="Changed", + body_storage="body", + ) + + +async def test_conflict_does_not_retry_write(monkeypatch): + calls = [] + + def handler(request): + calls.append(request.method) + return ( + httpx.Response(200, json=page()) + if request.method == "GET" + else httpx.Response(409, json={"message": "synthetic-confluence-access"}) + ) + + with pytest.raises(InvalidRequestError) as caught: + await run( + monkeypatch, + ConfluenceUpdatePageComponent, + "update_page", + handler, + space_id="10", + page_id="1", + expected_version=1, + title="Changed", + body_storage="body", + ) + assert calls == ["GET", "PUT"] + assert "synthetic-confluence-access" not in str(caught.value) + + +@pytest.mark.parametrize( + "path", + [ + "https://evil.invalid/", + "//evil.invalid", + "/wiki/api/v2/pages/../1", + "/wiki/api/v2/pages/1?space=20", + "/wiki/api/v2/pages/1%2f", + f"/ex/confluence/{SITE}/wiki/api/v2/pages", + "/wiki/api/v2/pages/1/attachments", + ], +) +async def test_arbitrary_targets_cannot_receive_credentials(path): + client = ConfluenceClient( + lease_for()[0], transport=httpx.MockTransport(lambda _r: pytest.fail("Unexpected request")) + ) + try: + with pytest.raises(InvalidRequestError): + await client.request("GET", path) + finally: + await client.aclose() + + +@pytest.mark.parametrize( + "changes", + [ + {"provider": "jira"}, + {"identity": "bot"}, + {"account": ConnectionAccount(id="user", tenant_id="other")}, + {"account": None}, + {"scopes_verified": False}, + {"granted_scopes": frozenset()}, + ], +) +async def test_credential_tenant_identity_and_scopes_fail_closed(changes): + client = ConfluenceClient( + lease_for(credential(**changes))[0], transport=httpx.MockTransport(lambda _r: pytest.fail("Unexpected request")) + ) + try: + with pytest.raises((ConnectionNotAuthorizedError, ScopeMissingError)): + await client.request("GET", "/wiki/api/v2/pages") + finally: + await client.aclose() + + +async def test_expired_access_token_refreshes_once_and_revalidates_tenant(): + lease, resolver = lease_for(credential(), credential(access_token=__import__("pydantic").SecretStr("replacement"))) + calls = [] + + def handler(request): + calls.append(request.headers["Authorization"]) + return httpx.Response(401) if len(calls) == 1 else httpx.Response(200, json={"results": []}) + + client = ConfluenceClient(lease, transport=httpx.MockTransport(handler)) + try: + assert await client.request("GET", "/wiki/api/v2/pages") == {"results": []} + assert calls == ["Bearer synthetic-confluence-access", "Bearer replacement"] + assert len(resolver.calls) == 2 + finally: + await client.aclose() + + +@pytest.mark.parametrize( + ("status", "error"), + [ + (401, AuthExpiredError), + (403, ConnectionNotAuthorizedError), + (404, ResourceNotFoundError), + (429, RateLimitedError), + (503, ProviderUnavailableError), + (302, ProviderUnavailableError), + ], +) +async def test_provider_errors_never_echo_tokens_or_provider_response(status, error): + client = ConfluenceClient( + lease_for()[0], + transport=httpx.MockTransport( + lambda _r: httpx.Response( + status, + text="synthetic-confluence-access", + headers={"Retry-After": "inf", "Location": "https://evil.invalid/"}, + ) + ), + ) + try: + with pytest.raises(error) as caught: + await client.request("GET", "/wiki/api/v2/pages") + assert "synthetic-confluence-access" not in str(caught.value) + finally: + await client.aclose() + + +async def test_network_errors_are_redacted(): + def handler(request): + message = "synthetic-confluence-access" + raise httpx.ConnectError(message, request=request) + + client = ConfluenceClient(lease_for()[0], transport=httpx.MockTransport(handler)) + try: + with pytest.raises(IntegrationNetworkError) as caught: + await client.request("GET", "/wiki/api/v2/pages") + assert "synthetic-confluence-access" not in str(caught.value) + finally: + await client.aclose() + + +@pytest.mark.parametrize( + "link", + [ + "https://evil.invalid/wiki/api/v2/pages?cursor=x", + "https://api.atlassian.com/ex/confluence/99999999-9999-9999-9999-999999999999/wiki/api/v2/pages?cursor=x", + "/wiki/api/v2/pages/1?cursor=x", + "/wiki/api/v2/pages?cursor=x&cursor=y", + "/wiki/api/v2/pages?other=x", + ], +) +def test_unsafe_pagination_links_rejected(link): + with pytest.raises(InvalidRequestError): + next_cursor({"_links": {"next": link}}) + + +@pytest.mark.parametrize("value", ["", "0", "../1", "10,20", "1?x=2", "https://evil.invalid", 10]) +def test_invalid_page_and_space_ids(value): + with pytest.raises(InvalidRequestError): + numeric_id(value) + + +def test_operator_scope_and_canonical_site_are_required(monkeypatch): + assert approved_space("10") == "10" + with pytest.raises(InvalidRequestError): + approved_space("30") + monkeypatch.setenv("LANGFLOW_CONFLUENCE_SPACE_IDS", "10,garbage") + with pytest.raises(InvalidRequestError): + approved_space("10") + monkeypatch.setenv("LANGFLOW_CONFLUENCE_CLOUD_ID", "https://evil.invalid") + with pytest.raises(InvalidRequestError): + cloud_id() + + +def test_manifest_and_standalone_extension_discovery_do_not_import_backend(): + class BlockBackend(importlib.abc.MetaPathFinder): + def find_spec(self, fullname, path=None, target=None): # noqa: ARG002 + if fullname == "langflow" or fullname.startswith("langflow."): + msg = "A standalone bundle imported the backend." + raise AssertionError(msg) + + blocker = BlockBackend() + sys.meta_path.insert(0, blocker) + try: + path = Path(__file__).parents[1] / "src/lfx_confluence" + validation = validate_extension(path) + assert validation.ok + loaded = load_extension(path) + assert loaded.ok + assert len(loaded.components) == 6 + assert {component.bundle for component in loaded.components} == {"confluence_integrations"} + assert len(load_manifest().capabilities) == 6 + component = ConfluenceOnPageCreatedTriggerComponent(connection="confluence/work", space_id="10") + assert component.trigger_config()["space_id"] == "10" + finally: + sys.meta_path.remove(blocker) + + +async def test_auth_refresh_cannot_rebind_another_tenant(): + lease, _ = lease_for(credential(), credential(account=ConnectionAccount(id="user-1", tenant_id="other"))) + calls = [] + + def handler(request): + calls.append(request) + return httpx.Response(401) + + client = ConfluenceClient(lease, transport=httpx.MockTransport(handler)) + try: + with pytest.raises(ConnectionNotAuthorizedError): + await client.request("GET", "/wiki/api/v2/pages") + assert len(calls) == 1 + finally: + await client.aclose() + + +async def test_oversized_and_malformed_responses_are_rejected(monkeypatch): + import lfx_confluence.client as module + + monkeypatch.setattr(module, "MAX_RESPONSE_BYTES", 32) + for content in ["x" * 33, "not-json"]: + client = ConfluenceClient( + lease_for()[0], transport=httpx.MockTransport(lambda _r, value=content: httpx.Response(200, text=value)) + ) + try: + with pytest.raises(ProviderUnavailableError): + await client.request("GET", "/wiki/api/v2/pages") + finally: + await client.aclose() diff --git a/src/frontend/src/components/core/triggers/__tests__/trigger-presentation.test.ts b/src/frontend/src/components/core/triggers/__tests__/trigger-presentation.test.ts index 88e5f3d0e03c..ee13dbaa4591 100644 --- a/src/frontend/src/components/core/triggers/__tests__/trigger-presentation.test.ts +++ b/src/frontend/src/components/core/triggers/__tests__/trigger-presentation.test.ts @@ -162,7 +162,7 @@ describe("sourceKeyOf", () => { ["salesforce", "Cloud"], ["servicenow", "Server"], ["jira", "Ticket"], - ["confluence", "BookOpen"], + ["confluence", "Confluence"], ])("shows the %s provider icon", (provider, icon) => { expect(sourceIconOf(makeTrigger({ provider }))).toBe(icon); }); diff --git a/src/frontend/src/components/core/triggers/helpers/trigger-presentation.ts b/src/frontend/src/components/core/triggers/helpers/trigger-presentation.ts index 9950ec83c34f..7ceaff054cb3 100644 --- a/src/frontend/src/components/core/triggers/helpers/trigger-presentation.ts +++ b/src/frontend/src/components/core/triggers/helpers/trigger-presentation.ts @@ -202,7 +202,7 @@ export const sourceIconOf = ( case "jira": return "Ticket"; case "confluence": - return "BookOpen"; + return "Confluence"; default: return trigger.kind === "inbound_webhook" ? "Webhook" : "Clock"; } diff --git a/src/frontend/src/pages/SettingsPage/pages/ConnectionsPage/__tests__/confluence-consent-scopes.test.ts b/src/frontend/src/pages/SettingsPage/pages/ConnectionsPage/__tests__/confluence-consent-scopes.test.ts new file mode 100644 index 000000000000..2d1313ee7eaa --- /dev/null +++ b/src/frontend/src/pages/SettingsPage/pages/ConnectionsPage/__tests__/confluence-consent-scopes.test.ts @@ -0,0 +1,37 @@ +import type { IntegrationCapabilityRead } from "@/controllers/API/queries/connections"; +import manifest from "../../../../../../../bundles/confluence/src/lfx_confluence/components/confluence_integrations/capabilities.v1.json"; +import scenarios from "../../../../../../../bundles/confluence/tests/fixtures/consent-scopes.json"; +import { + defaultScopesForNewConnection, + scopeRequirements, + scopesForRegistration, + uniqueScopes, +} from "../helpers/scopes"; + +it.each(scenarios)( + "requests usable confluence consent under $policy policy", + ({ policy, scopes }) => { + const capabilities = manifest.capabilities + .filter((capability) => policy === "all" || capability.risk === policy) + .map((capability) => ({ + ...capability, + allowed: true, + })) as IntegrationCapabilityRead[]; + const componentScopes = uniqueScopes( + scopeRequirements(capabilities, undefined), + ); + const { requestable, unavailable } = scopesForRegistration( + "confluence", + componentScopes, + manifest.auth_profiles[0].default_scopes, + ); + expect(unavailable).toEqual([]); + expect( + defaultScopesForNewConnection( + "confluence", + componentScopes, + requestable, + ).sort(), + ).toEqual(scopes); + }, +); diff --git a/src/frontend/src/utils/__tests__/sidebarBundles.test.ts b/src/frontend/src/utils/__tests__/sidebarBundles.test.ts index d460361841ca..267a4336fa80 100644 --- a/src/frontend/src/utils/__tests__/sidebarBundles.test.ts +++ b/src/frontend/src/utils/__tests__/sidebarBundles.test.ts @@ -103,3 +103,11 @@ describe("SIDEBAR_BUNDLES", () => { ); }); }); + +it("uses the Confluence logo for its integration palette group", () => { + expect(SIDEBAR_BUNDLES).toContainEqual({ + display_name: "Confluence", + name: "confluence_integrations", + icon: "Confluence", + }); +}); diff --git a/src/frontend/src/utils/styleUtils.ts b/src/frontend/src/utils/styleUtils.ts index 121ccfe7d98b..4ee62f0efc29 100644 --- a/src/frontend/src/utils/styleUtils.ts +++ b/src/frontend/src/utils/styleUtils.ts @@ -455,6 +455,11 @@ export const SIDEBAR_BUNDLES = [ { display_name: "CometAPI", name: "cometapi", icon: "CometAPI" }, { display_name: "Composio", name: "composio", icon: "Composio" }, { display_name: "Confluence", name: "confluence", icon: "Confluence" }, + { + display_name: "Confluence", + name: "confluence_integrations", + icon: "Confluence", + }, { display_name: "Couchbase", name: "couchbase", icon: "Couchbase" }, { display_name: "CrewAI", name: "crewai", icon: "CrewAI" }, { display_name: "CUGA", name: "cuga", icon: "Bot" }, diff --git a/src/lfx/src/lfx/cli/validation/_env_validation.py b/src/lfx/src/lfx/cli/validation/_env_validation.py index 7535d375d4b1..f92b68c7d401 100644 --- a/src/lfx/src/lfx/cli/validation/_env_validation.py +++ b/src/lfx/src/lfx/cli/validation/_env_validation.py @@ -7,7 +7,7 @@ if TYPE_CHECKING: from lfx.graph.graph.base import Graph - from lfx.integrations.errors import ConnectionUnresolvedError + from lfx.integrations.errors import IntegrationError def is_valid_env_var_name(name: str) -> bool: @@ -75,12 +75,13 @@ def validate_global_variables_for_env(graph: Graph) -> list[str]: return errors -def validate_connection_refs_for_env(graph: Graph) -> list[ConnectionUnresolvedError]: +def validate_connection_refs_for_env(graph: Graph) -> list[IntegrationError]: """Return typed failures for connection refs absent from headless injection channels.""" - from lfx.integrations.errors import ConnectionUnresolvedError + from lfx.integrations.errors import ConnectionUnresolvedError, IntegrationContextUnsupportedError from lfx.integrations.models import ConnectionRef from lfx.services.connection.env_resolver import EnvConnectionResolver from lfx.services.deps import get_connection_resolver + from lfx.services.integration_policy.utils import require_headless_integration_support from lfx.services.variable.request_scope import ( get_active_request_variables, is_env_fallback_disabled, @@ -88,7 +89,7 @@ def validate_connection_refs_for_env(graph: Graph) -> list[ConnectionUnresolvedE ) from lfx.utils.env_var_security import safe_getenv - errors: list[ConnectionUnresolvedError] = [] + errors: list[IntegrationError] = [] request_variables = normalize_parsed_variables( graph.context.get("request_variables") or get_active_request_variables() or {} ) @@ -100,6 +101,13 @@ def validate_connection_refs_for_env(graph: Graph) -> list[ConnectionUnresolvedE for field_name, field in template.items(): if not isinstance(field, dict) or field.get("type") != "connection_ref": continue + principal = getattr(graph, "execution_principal", None) + if principal is not None and principal.kind == "headless_operator": + try: + require_headless_integration_support(field.get("provider", ""), field.get("capabilities", ())) + except IntegrationContextUnsupportedError as exc: + errors.append(exc) + continue value = vertex.params.get(field_name, field.get("value")) if not value: continue diff --git a/src/lfx/src/lfx/custom/custom_component/component.py b/src/lfx/src/lfx/custom/custom_component/component.py index 64fa8083cafb..5ddf9eed405c 100644 --- a/src/lfx/src/lfx/custom/custom_component/component.py +++ b/src/lfx/src/lfx/custom/custom_component/component.py @@ -1624,6 +1624,7 @@ def _require_integration_policy_for_input(self, input_model, purpose=None) -> tu effective_purpose = purpose or IntegrationPolicyPurpose.USE capability_ids = self._selected_integration_capabilities(input_model.provider, input_model.capabilities) + self._require_headless_integration_support(input_model.provider, capability_ids) require_integration_actions( user_id=self.user_id, provider_id=input_model.provider, @@ -1633,6 +1634,14 @@ def _require_integration_policy_for_input(self, input_model, purpose=None) -> tu ) return capability_ids + def _require_headless_integration_support(self, provider_id: str, capability_ids: Iterable[str]) -> None: + from lfx.services.integration_policy.utils import require_headless_integration_support + + graph = getattr(self, "graph", None) + principal = getattr(graph, "execution_principal", None) + if principal is not None and principal.kind == "headless_operator": + require_headless_integration_support(provider_id, capability_ids) + def require_integration_policy(self, purpose=None) -> None: """Gate the selected integration actions before any adapter can run. @@ -1655,6 +1664,7 @@ def require_integration_policy(self, purpose=None) -> None: for provider_id, capabilities in integration_capabilities_for_component_class(type(self).__name__).items(): selected = self._selected_integration_capabilities(provider_id, capabilities) + self._require_headless_integration_support(provider_id, selected) require_integration_actions( user_id=self.user_id, provider_id=provider_id, diff --git a/src/lfx/src/lfx/integrations/errors.py b/src/lfx/src/lfx/integrations/errors.py index 07c6ff9995c8..d02e4556e3c9 100644 --- a/src/lfx/src/lfx/integrations/errors.py +++ b/src/lfx/src/lfx/integrations/errors.py @@ -372,6 +372,20 @@ def __init__(self, *, provider: str | None = None, http_status: int | None = Non ) +class IntegrationContextUnsupportedError(IntegrationError): + """A declared provider action cannot run in this execution environment.""" + + code = "action-unsupported" + + def __init__(self, *, provider: str, display_name: str) -> None: + super().__init__( + f"{display_name} connections are not available in headless lfx. " + "Run this flow on a supported Langflow server.", + provider=provider, + details={"context": "headless"}, + ) + + class IncompatibleToolError(IntegrationError): """A pinned MCP server no longer matches the tool contract a bundle pinned. diff --git a/src/lfx/src/lfx/services/integration_policy/utils.py b/src/lfx/src/lfx/services/integration_policy/utils.py index 863d4c1d6054..d0199bede57c 100644 --- a/src/lfx/src/lfx/services/integration_policy/utils.py +++ b/src/lfx/src/lfx/services/integration_policy/utils.py @@ -149,6 +149,25 @@ def integration_capabilities_for_component_class(class_name: str) -> dict[str, t } +def require_headless_integration_support(provider_id: str, capability_ids: Iterable[str]) -> None: + """Reject explicitly unsupported actions before suggesting credential setup.""" + from lfx.extension.bundle_registry import get_default_registry + from lfx.integrations.errors import IntegrationContextUnsupportedError + + selected = set(capability_ids) + if not selected: + return + for integration in get_default_registry().list_integrations(): + if integration.provider_id != provider_id: + continue + manifest = integration.capability_manifest + if any( + capability.id in selected and "headless" not in capability.deployment_contexts + for capability in manifest.capabilities + ): + raise IntegrationContextUnsupportedError(provider=provider_id, display_name=manifest.display_name) + + def integration_policy_identity_for_component_class(class_name: str) -> tuple[str, tuple[str, ...]] | None: """Return ``(provider_id, policy_keys)`` for the class a capability names. diff --git a/src/lfx/tests/unit/services/integration_policy/test_headless_support.py b/src/lfx/tests/unit/services/integration_policy/test_headless_support.py new file mode 100644 index 000000000000..3ed949c4ec1f --- /dev/null +++ b/src/lfx/tests/unit/services/integration_policy/test_headless_support.py @@ -0,0 +1,113 @@ +"""Unsupported integrations fail before headless credential setup or execution.""" + +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest +from lfx.cli.validation import validate_connection_refs_for_env +from lfx.custom.custom_component.component import Component +from lfx.integrations.errors import IntegrationContextUnsupportedError +from lfx.io import ConnectionRefInput, Output +from lfx.services.authorization.base import ExecutionPrincipal + + +class JiraProbe(Component): + inputs = [ConnectionRefInput(name="connection", provider="jira", capabilities=["jira.issue.read"])] + outputs = [Output(name="result", display_name="Result", method="read_issue")] + + def read_issue(self): + pytest.fail("Unsupported action must not run") + + +@pytest.fixture +def registry(monkeypatch): + capability = SimpleNamespace( + id="jira.issue.read", + component_ref="JiraProbe", + policy_keys=("integrations.jira.issue.read",), + deployment_contexts=("hosted", "self_managed"), + ) + manifest = SimpleNamespace(display_name="Jira Cloud", capabilities=[capability]) + monkeypatch.setattr( + "lfx.extension.bundle_registry.get_default_registry", + lambda: SimpleNamespace( + list_integrations=lambda: [SimpleNamespace(provider_id="jira", capability_manifest=manifest)] + ), + ) + return capability + + +def component(*, headless=True): + node = JiraProbe(connection="jira/work") + node.set_vertex( + SimpleNamespace( + outgoing_edges=[], + graph=SimpleNamespace( + execution_principal=ExecutionPrincipal(kind="headless_operator" if headless else "actor"), + flow_id="flow", + session_id="session", + ), + ) + ) + return node + + +@pytest.mark.usefixtures("registry") +def test_unsupported_context_precedes_resolver(monkeypatch): + monkeypatch.setattr("lfx.services.deps.get_connection_resolver", lambda: pytest.fail("Resolved credentials")) + with pytest.raises(IntegrationContextUnsupportedError, match=r"Jira Cloud.*not available in headless lfx"): + component().resolve_connection("connection") + + +@pytest.mark.usefixtures("registry") +async def test_unsupported_context_precedes_component_body(monkeypatch): + node = component() + send_message = AsyncMock() + monkeypatch.setattr(node, "send_message", send_message) + with pytest.raises(IntegrationContextUnsupportedError): + await node.build_results() + assert "action-unsupported" in send_message.call_args.args[0].model_dump_json() + + +@pytest.mark.parametrize("injected", [False, True]) +@pytest.mark.usefixtures("registry") +def test_cli_context_error_precedes_environment_setup(monkeypatch, *, injected): + if injected: + monkeypatch.setenv("LF_CONNECTION__JIRA__WORK", "synthetic-token") + else: + monkeypatch.delenv("LF_CONNECTION__JIRA__WORK", raising=False) + graph = SimpleNamespace( + execution_principal=ExecutionPrincipal(kind="headless_operator"), + context={}, + vertices=[ + SimpleNamespace( + params={}, + data={ + "node": { + "template": { + "connection": { + "type": "connection_ref", + "provider": "jira", + "capabilities": ["jira.issue.read"], + "value": "jira/work", + } + } + } + }, + ) + ], + ) + errors = validate_connection_refs_for_env(graph) + assert len(errors) == 1 + assert isinstance(errors[0], IntegrationContextUnsupportedError) + assert "LF_CONNECTION" not in str(errors[0]) + + +@pytest.mark.parametrize("supported", [False, True]) +def test_hosted_execution_and_supported_headless_remain_available(monkeypatch, registry, *, supported): + if supported: + registry.deployment_contexts = ("headless",) + resolver = SimpleNamespace() + monkeypatch.setattr("lfx.services.deps.get_connection_resolver", lambda: resolver) + lease = component(headless=supported).resolve_connection("connection") + assert lease is not None diff --git a/uv.lock b/uv.lock index e031d3d4c5e4..29db884a16c4 100644 --- a/uv.lock +++ b/uv.lock @@ -47,6 +47,7 @@ members = [ "lfx-bundles", "lfx-cohere", "lfx-confluent", + "lfx-confluence", "lfx-datastax", "lfx-docling", "lfx-duckduckgo", @@ -8915,6 +8916,7 @@ dependencies = [ { name = "lfx-anthropic" }, { name = "lfx-azure" }, { name = "lfx-cohere" }, + { name = "lfx-confluence" }, { name = "lfx-datastax" }, { name = "lfx-docling" }, { name = "lfx-github" }, @@ -9050,6 +9052,7 @@ requires-dist = [ { name = "lfx-azure", editable = "src/bundles/azure" }, { name = "lfx-bundles", extras = ["all-no-torch"], marker = "extra == 'bundles'", editable = "src/bundles/lfx-bundles" }, { name = "lfx-cohere", editable = "src/bundles/cohere" }, + { name = "lfx-confluence", editable = "src/bundles/confluence" }, { name = "lfx-confluent", marker = "extra == 'bundles'", editable = "src/bundles/confluent" }, { name = "lfx-datastax", editable = "src/bundles/datastax" }, { name = "lfx-docling", editable = "src/bundles/docling" }, @@ -11125,6 +11128,21 @@ requires-dist = [ { name = "pyiceberg", specifier = ">=0.10,<1" }, ] +[[package]] +name = "lfx-confluence" +version = "0.1.0" +source = { editable = "src/bundles/confluence" } +dependencies = [ + { name = "httpx" }, + { name = "lfx" }, +] + +[package.metadata] +requires-dist = [ + { name = "httpx", specifier = ">=0.27,<1" }, + { name = "lfx", editable = "src/lfx" }, +] + [[package]] name = "lfx-datastax" version = "0.2.1"