diff --git a/design/dedicated-integrations/connection-contract.md b/design/dedicated-integrations/connection-contract.md index 6258b0b88dbc..89da16875e5c 100644 --- a/design/dedicated-integrations/connection-contract.md +++ b/design/dedicated-integrations/connection-contract.md @@ -5,7 +5,7 @@ Decision ID: connection-contract Applies to: INT-2 (lfx), with the langflow-base obligations INT-4 and INT-5 must meet and the Enterprise seams Owners (sign-off roles): lfx owner, langflow-base owner, Enterprise owner, frontend owner Last verified: 2026-09-01 -Last amended: 2026-09-03 (INT-2 implementation review) +Last amended: 2026-09-03 (connection persistence implementation review) This document is the INT-2 design that the discovery gate asks the lfx, langflow-base, and Enterprise owners to sign off before INT-2 is built. Each section states the recommended decision, why, and what was rejected. Section 12 @@ -78,8 +78,8 @@ authorization. account identity is ever stored in flow JSON. - `required_connections`: the deployment artifact manifest adds `required_connections: [{provider, name, scopes}]` per flow and aggregated, beside `required_variables` - (`src/backend/base/langflow/services/deployment_artifacts/builder.py:347`); scopes come from the input's declared - `required_scopes`. The manifest `schema_version` bump is an INT-4 decision. + (`src/backend/base/langflow/services/deployment_artifacts/builder.py`); scopes come from the input's declared + `required_scopes`. Artifacts with connection requirements use manifest `schema_version: 4`. Rejected: an opaque connection UUID (does not survive export or `lfx run`, and the UI needs a lookup to display it); a `SecretStrInput` subclass with a synthetic variable name (password rendering, `load_from_db=True` from @@ -385,9 +385,9 @@ identifiers.** 3. Per-connection `allow_non_interactive` semantics, including `mcp_projects` with auth `none`. 4. Cross-worker single-flight refresh: a DB lease column versus a Redis lock; the `background_execution` lease-claim code is the precedent. -5. Encryption envelope: extend the `sso_secret.py` HKDF scheme with a new info label, or the Fernet - `encrypt_api_key` path used by MCP and variables. -6. Artifact manifest schema version for `required_connections`. +5. Encryption envelope decision: use the existing Fernet `encrypt_api_key` path used by MCP and variables, and + isolate the encrypted envelope in `connection_secret` so metadata queries never load credential material. +6. Artifact manifest schema decision: use version 4 when `required_connections` is non-empty. 7. Desktop: the same `GET /api/v1/connections/{provider}/callback` on `localhost:7860` with a PKCE public client that is Langflow-owned by default and customer-owned as the override (`decisions/desktop-oauth-ownership.md`), and the redirect allowlist (`127.0.0.1` loopback; Microsoft ignores the port when matching localhost redirects). diff --git a/docs/docs/API-Reference/api-reference-api-examples.mdx b/docs/docs/API-Reference/api-reference-api-examples.mdx index f3bc5b13b647..4bdd4b1d4de8 100644 --- a/docs/docs/API-Reference/api-reference-api-examples.mdx +++ b/docs/docs/API-Reference/api-reference-api-examples.mdx @@ -377,6 +377,16 @@ The following endpoints are most often used when contributing to the Langflow co * PATCH `/v1/variables/{variable_id}`: Update a variable. * DELETE `/v1/variables/{variable_id}`: Delete a variable. +* Connections: + * GET `/v1/connections`: List owned, instance-owned, and explicitly shared connection metadata. Use the optional `provider` query parameter to filter the list for a connection picker. + * POST `/v1/connections`: Create connection metadata and optionally provide credentials for encrypted storage. + * POST `/v1/connections/{connection_id}/test`: Test the stored credential and required scope coverage. + * POST `/v1/connections/{connection_id}/health`: Refresh the stored connection health. + * POST `/v1/connections/{connection_id}/revoke`: Remove stored credentials and mark the connection revoked. + * DELETE `/v1/connections/{connection_id}`: Delete a connection. + + Connection responses contain metadata, status, health, granted scopes, and a `has_credentials` flag. They never contain access tokens, refresh tokens, or encrypted credential payloads. User-owned connections can be used non-interactively only when `allow_non_interactive` is enabled; connections cannot be shared publicly. + * [Use voice mode](/concepts-voice-mode): * WS `/v1/voice/ws/flow_tts/{flow_id}`: Speech-to-text session that runs a flow and returns TTS. * WS `/v1/voice/ws/flow_tts/{flow_id}/{session_id}`: Same as above with explicit session ID. diff --git a/docs/docs/Develop/memory.mdx b/docs/docs/Develop/memory.mdx index 4e78ca61078a..9fbe241de750 100644 --- a/docs/docs/Develop/memory.mdx +++ b/docs/docs/Develop/memory.mdx @@ -49,6 +49,8 @@ On OSS, the default is the SQLite file `langflow.db`. Desktop uses `database.db` • **DeploymentProviderAccount**: Stores configured connections to external deployment providers, including provider URL, tenant settings, and encrypted credentials. +• **Connection** and **ConnectionSecret**: Store integration connection metadata separately from Fernet-encrypted credential payloads. Connection records carry ownership, provider, granted scopes, executing identity, health, and non-interactive-use policy; API reads never join or return the secret payload. + • **File**: Stores metadata for files uploaded to Langflow's file management system, including file names, paths, sizes, and storage providers. For more information, see [Manage files](/concepts-file-management). • **Flow**: Contains flow definitions, including nodes, edges, and components, stored as JSON or database records. For more information, see [Build flows](/concepts-flows). @@ -220,4 +222,4 @@ For more information and examples, see [**Message History** component](/message- * [Configure an external PostgreSQL database](/configuration-custom-database) * [Langflow file management](/concepts-file-management) * [Langflow logs](/logging) -* [Langflow environment variables](/environment-variables) \ No newline at end of file +* [Langflow environment variables](/environment-variables) diff --git a/scripts/ci/authz_endpoint_matrix.json b/scripts/ci/authz_endpoint_matrix.json index a950b7078af7..40a95e18ad89 100644 --- a/scripts/ci/authz_endpoint_matrix.json +++ b/scripts/ci/authz_endpoint_matrix.json @@ -437,6 +437,28 @@ "GET|/{job_id}/events|reattach_workflow_events|read|authenticated" ] }, + { + "family": "connections", + "source": "api/v1/connections.py", + "resource": "connection", + "domain": "caller-global owner or explicit share scope", + "privacy": "connection identifier denials are returned as 404; responses contain metadata and health only", + "side_effects": "authorize before credential decryption, health checks, revocation, or deletion", + "frontend": "connection pickers list only owned, instance, or explicitly shared metadata and never receive token material", + "personas": "canonical_v1", + "test_references": [ + "src/backend/tests/unit/api/v1/test_connections.py::test_non_owner_cannot_test_or_delete_connection", + "src/backend/tests/unit/api/v1/test_connections.py::test_connection_responses_never_include_tokens" + ], + "routes": [ + "GET||list_connections|read|authenticated", + "POST||create_connection|create|authenticated", + "POST|/{connection_id}/test|test_connection|execute|authenticated", + "POST|/{connection_id}/health|refresh_connection_health|execute|authenticated", + "POST|/{connection_id}/revoke|revoke_connection|write|authenticated", + "DELETE|/{connection_id}|delete_connection|delete|authenticated" + ] + }, { "family": "workflow_public_v2", "source": "api/v2/workflow_public.py", diff --git a/src/backend/base/langflow/alembic/versions/f3b6a9d2e4c1_add_connections.py b/src/backend/base/langflow/alembic/versions/f3b6a9d2e4c1_add_connections.py new file mode 100644 index 000000000000..fa4617eef156 --- /dev/null +++ b/src/backend/base/langflow/alembic/versions/f3b6a9d2e4c1_add_connections.py @@ -0,0 +1,103 @@ +"""Add connection metadata and encrypted credential tables. + +Revision ID: f3b6a9d2e4c1 +Revises: c9f2e5a7b1d4 +Create Date: 2026-09-03 + +Phase: EXPAND +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +import sqlalchemy as sa +from alembic import op +from langflow.utils import migration + +if TYPE_CHECKING: + from collections.abc import Sequence + +revision: str = "f3b6a9d2e4c1" # pragma: allowlist secret +down_revision: str | None = "c9f2e5a7b1d4" # pragma: allowlist secret +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + +CONNECTION_TABLE = "connection" +SECRET_TABLE = "connection_secret" # noqa: S105 # pragma: allowlist secret - table name + + +def upgrade() -> None: + conn = op.get_bind() + if not migration.table_exists(CONNECTION_TABLE, conn): + op.create_table( + CONNECTION_TABLE, + sa.Column("id", sa.Uuid(), nullable=False), + sa.Column("owner_id", sa.Uuid(), nullable=True), + sa.Column("provider_key", sa.String(length=120), nullable=False), + sa.Column("name", sa.String(length=64), nullable=False), + sa.Column("display_name", sa.String(length=255), nullable=False), + sa.Column("ownership_mode", sa.String(length=16), nullable=False), + sa.Column("status", sa.String(length=16), nullable=False), + sa.Column("health", sa.String(length=16), nullable=False), + sa.Column("granted_scopes", sa.JSON(), nullable=False), + sa.Column("executing_identity", sa.JSON(), nullable=False), + sa.Column("allow_non_interactive", sa.Boolean(), nullable=False, server_default=sa.false()), + sa.Column("health_checked_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), + sa.CheckConstraint( + "(ownership_mode = 'user' AND owner_id IS NOT NULL) OR " + "(ownership_mode = 'instance' AND owner_id IS NULL)", + name="ck_connection_owner_mode", + ), + sa.CheckConstraint( + "status IN ('pending', 'ready', 'expired', 'revoked', 'error')", + name="ck_connection_status", + ), + sa.CheckConstraint( + "health IN ('unknown', 'healthy', 'unhealthy')", + name="ck_connection_health", + ), + sa.ForeignKeyConstraint(["owner_id"], ["user.id"], ondelete="CASCADE"), + sa.PrimaryKeyConstraint("id"), + ) + op.create_index("ix_connection_owner_id", CONNECTION_TABLE, ["owner_id"], unique=False) + op.create_index( + "uq_connection_user_provider_name", + CONNECTION_TABLE, + ["owner_id", "provider_key", "name"], + unique=True, + sqlite_where=sa.text("ownership_mode = 'user'"), + postgresql_where=sa.text("ownership_mode = 'user'"), + ) + op.create_index( + "uq_connection_instance_provider_name", + CONNECTION_TABLE, + ["provider_key", "name"], + unique=True, + sqlite_where=sa.text("ownership_mode = 'instance'"), + postgresql_where=sa.text("ownership_mode = 'instance'"), + ) + + if not migration.table_exists(SECRET_TABLE, conn): + op.create_table( + SECRET_TABLE, + sa.Column("connection_id", sa.Uuid(), nullable=False), + sa.Column("encrypted_payload", sa.Text(), nullable=False), + sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), + sa.ForeignKeyConstraint(["connection_id"], ["connection.id"], ondelete="CASCADE"), + sa.PrimaryKeyConstraint("connection_id"), + ) + + +def downgrade() -> None: + conn = op.get_bind() + if migration.table_exists(SECRET_TABLE, conn): + op.drop_table(SECRET_TABLE) + if migration.table_exists(CONNECTION_TABLE, conn): + op.drop_index("uq_connection_instance_provider_name", table_name=CONNECTION_TABLE) + op.drop_index("uq_connection_user_provider_name", table_name=CONNECTION_TABLE) + op.drop_index("ix_connection_owner_id", table_name=CONNECTION_TABLE) + op.drop_table(CONNECTION_TABLE) diff --git a/src/backend/base/langflow/api/router.py b/src/backend/base/langflow/api/router.py index c35dde2920d6..2486af8afce3 100644 --- a/src/backend/base/langflow/api/router.py +++ b/src/backend/base/langflow/api/router.py @@ -17,6 +17,7 @@ authz_teams_router, catalog_policy_router, chat_router, + connections_router, endpoints_router, extensions_router, files_router, @@ -69,6 +70,7 @@ def include_deployment_router(target_router: APIRouter) -> None: router_v1.include_router(chat_router) +router_v1.include_router(connections_router) router_v1.include_router(endpoints_router) router_v1.include_router(validate_router) router_v1.include_router(store_router) diff --git a/src/backend/base/langflow/api/v1/__init__.py b/src/backend/base/langflow/api/v1/__init__.py index 27e476943bbc..39d3c0211164 100644 --- a/src/backend/base/langflow/api/v1/__init__.py +++ b/src/backend/base/langflow/api/v1/__init__.py @@ -9,6 +9,7 @@ from langflow.api.v1.authz_teams import router as authz_teams_router from langflow.api.v1.catalog_policy import router as catalog_policy_router from langflow.api.v1.chat import router as chat_router +from langflow.api.v1.connections import router as connections_router from langflow.api.v1.endpoints import router as endpoints_router from langflow.api.v1.extensions import router as extensions_router from langflow.api.v1.files import router as files_router @@ -48,6 +49,7 @@ "authz_teams_router", "catalog_policy_router", "chat_router", + "connections_router", "endpoints_router", "extensions_router", "files_router", diff --git a/src/backend/base/langflow/api/v1/authz_shares.py b/src/backend/base/langflow/api/v1/authz_shares.py index 72c523be9705..31284085bac3 100644 --- a/src/backend/base/langflow/api/v1/authz_shares.py +++ b/src/backend/base/langflow/api/v1/authz_shares.py @@ -26,6 +26,7 @@ SharePermissionLevel, ShareScope, ) +from langflow.services.database.models.connection import Connection from langflow.services.database.models.deployment.model import Deployment from langflow.services.database.models.file.model import File as UserFile from langflow.services.database.models.flow.model import Flow @@ -49,6 +50,7 @@ "knowledge_base": (KnowledgeBaseRecord, "user_id"), "variable": (Variable, "user_id"), "file": (UserFile, "user_id"), + "connection": (Connection, "owner_id"), } @@ -291,6 +293,11 @@ async def _ensure_can_administer_share( def _ensure_supported_share_permission(*, resource_type: str, scope: str, permission_level: str) -> None: """Reject share levels that have no matching public flow product behavior.""" + if resource_type == "connection" and scope == ShareScope.PUBLIC.value: + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, + detail="Connections cannot be shared publicly.", + ) if resource_type != "flow" or scope != ShareScope.PUBLIC.value: return if permission_level == SharePermissionLevel.EXECUTE.value: diff --git a/src/backend/base/langflow/api/v1/connections.py b/src/backend/base/langflow/api/v1/connections.py new file mode 100644 index 000000000000..150337ee73f4 --- /dev/null +++ b/src/backend/base/langflow/api/v1/connections.py @@ -0,0 +1,189 @@ +"""Owner- and share-aware API for persisted integration connections.""" + +from __future__ import annotations + +from typing import Annotated +from uuid import UUID + +from fastapi import APIRouter, Depends, HTTPException, Query, Response, status +from lfx.integrations.models import PROVIDER_ID_PATTERN +from lfx.services.authorization.base import ExecutionPrincipal + +from langflow.api.utils import CurrentActiveUser, DbSession, DbSessionReadOnly +from langflow.services.authorization import ConnectionAction, ensure_connection_permission +from langflow.services.connection import ConnectionConflictError, DatabaseConnectionResolverService +from langflow.services.database.models.connection import ( + Connection, + ConnectionCreate, + ConnectionRead, + ConnectionTestRequest, +) +from langflow.services.deps import get_connection_resolver_service + +router = APIRouter(prefix="/connections", tags=["Connections"]) + + +def _database_service() -> DatabaseConnectionResolverService: + service = get_connection_resolver_service() + if not isinstance(service, DatabaseConnectionResolverService): + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail="Connection metadata is managed by the configured host service.", + ) + return service + + +ConnectionService = Annotated[DatabaseConnectionResolverService, Depends(_database_service)] + + +def _interactive_principal(user: CurrentActiveUser) -> ExecutionPrincipal: + return ExecutionPrincipal( + kind="actor", + user_id=str(user.id), + actor_id=str(user.id), + family="connections_api", + interactive=True, + actor_label=user.username, + ) + + +async def _authorized_row( + *, + service: DatabaseConnectionResolverService, + session: DbSession | DbSessionReadOnly, + user: CurrentActiveUser, + connection_id: UUID, + action: ConnectionAction, + for_update: bool = False, +) -> Connection: + row = await service.get_for_user(session, user=user, connection_id=connection_id, for_update=for_update) + if row is None: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Connection not found") + await ensure_connection_permission( + user, + action, + connection_id=row.id, + connection_owner_id=row.owner_id, + ) + return row + + +@router.get("", response_model=list[ConnectionRead]) +async def list_connections( + session: DbSessionReadOnly, + current_user: CurrentActiveUser, + service: ConnectionService, + provider: Annotated[str | None, Query(pattern=PROVIDER_ID_PATTERN, max_length=120)] = None, +) -> list[ConnectionRead]: + """List owned, instance-owned, and explicitly shared connection metadata.""" + return await service.list_for_user(session, user=current_user, provider_key=provider) + + +@router.post("", response_model=ConnectionRead, status_code=status.HTTP_201_CREATED) +async def create_connection( + payload: ConnectionCreate, + session: DbSession, + current_user: CurrentActiveUser, + service: ConnectionService, +) -> ConnectionRead: + """Create connection metadata and optionally store encrypted credentials.""" + if payload.ownership_mode.value == "instance" and not current_user.is_superuser: + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail="Only a superuser may create an instance connection.", + ) + await ensure_connection_permission( + current_user, + ConnectionAction.CREATE, + connection_owner_id=current_user.id, + ) + try: + return await service.create(session, user=current_user, payload=payload) + except ConnectionConflictError as exc: + raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(exc)) from exc + + +@router.post("/{connection_id}/test", response_model=ConnectionRead) +async def test_connection( + connection_id: UUID, + payload: ConnectionTestRequest, + session: DbSession, + current_user: CurrentActiveUser, + service: ConnectionService, +) -> ConnectionRead: + """Validate the local credential envelope and requested scope coverage.""" + row = await _authorized_row( + service=service, + session=session, + user=current_user, + connection_id=connection_id, + action=ConnectionAction.EXECUTE, + for_update=True, + ) + return await service.check_health( + session, + row=row, + principal=_interactive_principal(current_user), + required_scopes=frozenset(payload.required_scopes), + ) + + +@router.post("/{connection_id}/health", response_model=ConnectionRead) +async def refresh_connection_health( + connection_id: UUID, + session: DbSession, + current_user: CurrentActiveUser, + service: ConnectionService, +) -> ConnectionRead: + """Refresh credential health without returning or logging token material.""" + row = await _authorized_row( + service=service, + session=session, + user=current_user, + connection_id=connection_id, + action=ConnectionAction.EXECUTE, + for_update=True, + ) + return await service.check_health( + session, + row=row, + principal=_interactive_principal(current_user), + ) + + +@router.post("/{connection_id}/revoke", response_model=ConnectionRead) +async def revoke_connection( + connection_id: UUID, + session: DbSession, + current_user: CurrentActiveUser, + service: ConnectionService, +) -> ConnectionRead: + """Remove local credential material and mark the connection revoked.""" + row = await _authorized_row( + service=service, + session=session, + user=current_user, + connection_id=connection_id, + action=ConnectionAction.WRITE, + for_update=True, + ) + return await service.revoke(session, row) + + +@router.delete("/{connection_id}", status_code=status.HTTP_204_NO_CONTENT) +async def delete_connection( + connection_id: UUID, + session: DbSession, + current_user: CurrentActiveUser, + service: ConnectionService, +) -> Response: + row = await _authorized_row( + service=service, + session=session, + user=current_user, + connection_id=connection_id, + action=ConnectionAction.DELETE, + for_update=True, + ) + await service.delete(session, row) + return Response(status_code=status.HTTP_204_NO_CONTENT) diff --git a/src/backend/base/langflow/api/v1/schemas/authz_shares.py b/src/backend/base/langflow/api/v1/schemas/authz_shares.py index c20adf3a9c02..bd41c9bce6ce 100644 --- a/src/backend/base/langflow/api/v1/schemas/authz_shares.py +++ b/src/backend/base/langflow/api/v1/schemas/authz_shares.py @@ -16,6 +16,7 @@ "knowledge_base", "variable", "file", + "connection", ] ShareScopeLiteral = Literal["private", "team", "user", "public"] diff --git a/src/backend/base/langflow/services/authorization/__init__.py b/src/backend/base/langflow/services/authorization/__init__.py index 914232f16178..d7cbacb16acb 100644 --- a/src/backend/base/langflow/services/authorization/__init__.py +++ b/src/backend/base/langflow/services/authorization/__init__.py @@ -1,6 +1,7 @@ """OSS authorization service package (pass-through default; plugins enforce).""" from langflow.services.authorization.actions import ( + ConnectionAction, DeploymentAction, FileAction, FlowAction, @@ -21,6 +22,7 @@ from langflow.services.authorization.fetch import authorized_or_owner_scoped, deny_to_404 from langflow.services.authorization.guards import ( capability_probe, + ensure_connection_permission, ensure_deployment_permission, ensure_file_permission, ensure_flow_permission, @@ -48,6 +50,7 @@ __all__ = [ "AuditPersistenceError", + "ConnectionAction", "DeploymentAction", "FileAction", "FlowAction", @@ -65,6 +68,7 @@ "capability_probe", "deny_to_404", "drain_pending_audit_writes", + "ensure_connection_permission", "ensure_deployment_permission", "ensure_file_permission", "ensure_flow_permission", diff --git a/src/backend/base/langflow/services/authorization/actions.py b/src/backend/base/langflow/services/authorization/actions.py index 77446a00ba16..ddccdc3f3d55 100644 --- a/src/backend/base/langflow/services/authorization/actions.py +++ b/src/backend/base/langflow/services/authorization/actions.py @@ -72,6 +72,16 @@ class ProviderAccountAction(str, Enum): DELETE = "delete" +class ConnectionAction(str, Enum): + """Actions authorized on a persisted integration connection.""" + + READ = "read" + WRITE = "write" + CREATE = "create" + DELETE = "delete" + EXECUTE = "execute" + + class VoiceAction(str, Enum): """Actions authorized on credential-backed voice resources.""" diff --git a/src/backend/base/langflow/services/authorization/guards.py b/src/backend/base/langflow/services/authorization/guards.py index b4f3e87f331e..2a61f38d5a20 100644 --- a/src/backend/base/langflow/services/authorization/guards.py +++ b/src/backend/base/langflow/services/authorization/guards.py @@ -22,6 +22,7 @@ get_current_external_access_context, ) from langflow.services.authorization.actions import ( + ConnectionAction, DeploymentAction, FileAction, FlowAction, @@ -50,6 +51,7 @@ FileAction, ShareAction, ProviderAccountAction, + ConnectionAction, VoiceAction, ) @@ -64,6 +66,7 @@ "file_user_id", "share_user_id", "provider_account_user_id", + "connection_owner_id", "voice_user_id", ) @@ -179,6 +182,7 @@ def _coerce_action( | FileAction | ShareAction | ProviderAccountAction + | ConnectionAction | VoiceAction | str, ) -> str: @@ -442,6 +446,16 @@ class _ResourceSpec: # A newly-created provider account always belongs to the caller. owner_override_on_create=True, ), + "connection": _ResourceSpec( + resource_type="connection", + owner_kw="connection_owner_id", + id_kw="connection_id", + workspace_kw=None, + scope_kw=None, + # User-owned creates belong to the caller. Instance-owned creates are + # additionally restricted to superusers by the API route. + owner_override_on_create=True, + ), "voice": _ResourceSpec( resource_type="voice", owner_kw="voice_user_id", @@ -760,6 +774,27 @@ async def ensure_provider_account_permission( ) +async def ensure_connection_permission( + user: User | UserRead, + act: ConnectionAction | str, + *, + connection_id: UUID | None = None, + connection_owner_id: UUID | None = None, + domain: str | None = None, +) -> None: + """Check permission for connection metadata or credential use.""" + await _ensure_typed( + user, + spec_key="connection", + act_str=_coerce_action(act), + kwargs={ + "connection_id": connection_id, + "connection_owner_id": connection_owner_id, + }, + domain_override=domain, + ) + + async def ensure_voice_permission( user: User | UserRead, act: VoiceAction | str, diff --git a/src/backend/base/langflow/services/authorization/permissions.py b/src/backend/base/langflow/services/authorization/permissions.py index d50ddedf1280..6ece962396e4 100644 --- a/src/backend/base/langflow/services/authorization/permissions.py +++ b/src/backend/base/langflow/services/authorization/permissions.py @@ -6,6 +6,7 @@ from langflow.services.authorization.actions import ( AdministrationAction, + ConnectionAction, DeploymentAction, FileAction, FlowAction, @@ -29,6 +30,7 @@ "file": frozenset({action.value for action in FileAction}) | {"*"}, "share": frozenset({action.value for action in ShareAction}) | {"*"}, "provider_account": frozenset({action.value for action in ProviderAccountAction}) | {"*"}, + "connection": frozenset({action.value for action in ConnectionAction}) | {"*"}, "voice": frozenset({action.value for action in VoiceAction}) | {"*"}, } diff --git a/src/backend/base/langflow/services/connection/__init__.py b/src/backend/base/langflow/services/connection/__init__.py new file mode 100644 index 000000000000..8059bfefb32b --- /dev/null +++ b/src/backend/base/langflow/services/connection/__init__.py @@ -0,0 +1,11 @@ +from .service import ( + ConnectionConflictError, + ConnectionSecretError, + DatabaseConnectionResolverService, +) + +__all__ = [ + "ConnectionConflictError", + "ConnectionSecretError", + "DatabaseConnectionResolverService", +] diff --git a/src/backend/base/langflow/services/connection/factory.py b/src/backend/base/langflow/services/connection/factory.py new file mode 100644 index 000000000000..5c17bbb34e69 --- /dev/null +++ b/src/backend/base/langflow/services/connection/factory.py @@ -0,0 +1,18 @@ +"""Factory for Langflow's database-backed connection resolver.""" + +from typing_extensions import override + +from langflow.services.connection.service import DatabaseConnectionResolverService +from langflow.services.factory import ServiceFactory +from langflow.services.schema import ServiceType + + +class ConnectionResolverServiceFactory(ServiceFactory): + name = ServiceType.CONNECTION_RESOLVER_SERVICE.value + + def __init__(self) -> None: + super().__init__(DatabaseConnectionResolverService) + + @override + def create(self) -> DatabaseConnectionResolverService: + return self.service_class() diff --git a/src/backend/base/langflow/services/connection/service.py b/src/backend/base/langflow/services/connection/service.py new file mode 100644 index 000000000000..e0493cda72f2 --- /dev/null +++ b/src/backend/base/langflow/services/connection/service.py @@ -0,0 +1,460 @@ +"""Database-backed connection persistence and runtime resolution.""" + +from __future__ import annotations + +import json +from datetime import datetime, timezone +from typing import TYPE_CHECKING +from uuid import UUID + +from lfx.integrations.errors import ( + AuthExpiredError, + ConnectionUnresolvedError, + IntegrationError, + ScopeMissingError, +) +from lfx.integrations.models import ( + ConnectionRef, + ConnectionResolutionRequest, + ConnectionStatus, + ResolvedCredential, +) +from lfx.services.connection.base import BaseConnectionResolverService +from pydantic import SecretStr +from sqlalchemy.exc import IntegrityError +from sqlmodel import col, or_, select + +from langflow.services.auth import utils as auth_utils +from langflow.services.authorization import filter_visible_resources, visible_scope_prefilter +from langflow.services.authorization.listing import apply_owned_or_visible_scope_prefilter +from langflow.services.database.models.connection import ( + Connection, + ConnectionCreate, + ConnectionHealth, + ConnectionOwnershipMode, + ConnectionRead, + ConnectionSecret, + ExecutingIdentityDescriptor, + PersistedConnectionStatus, +) +from langflow.services.deps import get_authorization_service, get_settings_service, session_scope + +if TYPE_CHECKING: + from lfx.services.authorization.base import ExecutionPrincipal + from sqlmodel.ext.asyncio.session import AsyncSession + + from langflow.services.database.models.user.model import User, UserRead + + +class ConnectionConflictError(ValueError): + """Raised when an owner already has a connection with the same handle.""" + + +class ConnectionSecretError(RuntimeError): + """Raised when connection credential material cannot be encrypted or decoded.""" + + +def _utc_now() -> datetime: + return datetime.now(timezone.utc) + + +def _credential_payload(payload: ConnectionCreate) -> str | None: + credentials = payload.credentials + if credentials is None: + return None + return json.dumps( + { + "version": 1, + "access_token": credentials.access_token.get_secret_value(), + "refresh_token": ( + credentials.refresh_token.get_secret_value() if credentials.refresh_token is not None else None + ), + "token_type": credentials.token_type, + "expires_at": credentials.expires_at.isoformat() if credentials.expires_at is not None else None, + }, + separators=(",", ":"), + sort_keys=True, + ) + + +def _encrypt_credential_payload(payload: str) -> str: + try: + return auth_utils.encrypt_api_key(payload) + except Exception as exc: + msg = "Connection credential encryption failed; check the server encryption configuration" + raise ConnectionSecretError(msg) from exc + + +def _decrypt_credential_payload(encrypted_payload: str) -> dict: + try: + plaintext = auth_utils.decrypt_api_key(encrypted_payload) + if not plaintext: + msg = "empty decrypted payload" + raise ValueError(msg) + decoded = json.loads(plaintext) + if not isinstance(decoded, dict) or decoded.get("version") != 1: + msg = "unsupported credential envelope" + raise ValueError(msg) + access_token = decoded.get("access_token") + if not isinstance(access_token, str) or not access_token: + msg = "credential envelope has no access token" + raise ValueError(msg) + except Exception as exc: + msg = "Stored connection credential could not be decoded" + raise ConnectionSecretError(msg) from exc + return decoded + + +def _parse_expiry(value: object) -> datetime | None: + if value is None: + return None + if not isinstance(value, str): + msg = "Stored connection credential has an invalid expiry" + raise ConnectionSecretError(msg) + try: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError as exc: + msg = "Stored connection credential has an invalid expiry" + raise ConnectionSecretError(msg) from exc + return parsed if parsed.tzinfo is not None else parsed.replace(tzinfo=timezone.utc) + + +class DatabaseConnectionResolverService(BaseConnectionResolverService): + """Resolve encrypted database connections while exposing only safe metadata.""" + + async def create( + self, + session: AsyncSession, + *, + user: User | UserRead, + payload: ConnectionCreate, + ) -> ConnectionRead: + owner_id = user.id if payload.ownership_mode == ConnectionOwnershipMode.USER else None + now = _utc_now() + raw_credentials = _credential_payload(payload) + encrypted_payload = _encrypt_credential_payload(raw_credentials) if raw_credentials is not None else None + row = Connection( + owner_id=owner_id, + ownership_mode=payload.ownership_mode.value, + provider_key=payload.provider_key, + name=payload.name, + display_name=payload.display_name, + status=( + PersistedConnectionStatus.READY.value + if encrypted_payload is not None + else PersistedConnectionStatus.PENDING.value + ), + health=ConnectionHealth.UNKNOWN.value, + granted_scopes=list(payload.granted_scopes), + executing_identity=payload.executing_identity.model_dump(mode="json"), + allow_non_interactive=payload.allow_non_interactive, + created_at=now, + updated_at=now, + ) + session.add(row) + try: + await session.flush() + if encrypted_payload is not None: + session.add(ConnectionSecret(connection_id=row.id, encrypted_payload=encrypted_payload)) + await session.flush() + except IntegrityError as exc: + await session.rollback() + msg = "A connection with this provider and name already exists" + raise ConnectionConflictError(msg) from exc + await session.refresh(row) + return self.to_read(row, has_credentials=encrypted_payload is not None) + + async def list_for_user( + self, + session: AsyncSession, + *, + user: User | UserRead, + provider_key: str | None = None, + ) -> list[ConnectionRead]: + is_superuser = bool(getattr(user, "is_superuser", False)) + owner_clause = or_( + Connection.owner_id == user.id, + Connection.ownership_mode == ConnectionOwnershipMode.INSTANCE.value, + ) + stmt = select(Connection) + if provider_key is not None: + stmt = stmt.where(Connection.provider_key == provider_key) + authz = get_authorization_service() + cross_user = await authz.is_enabled() and await authz.supports_cross_user_fetch() + if not is_superuser: + if cross_user: + visibility = await visible_scope_prefilter(user, resource_type="connection", act="read") + if visibility is not None: + stmt = await apply_owned_or_visible_scope_prefilter( + stmt, + id_column=Connection.id, + owner_clause=owner_clause, + visibility=visibility, + ) + else: + stmt = stmt.where(owner_clause) + stmt = stmt.order_by(col(Connection.display_name), col(Connection.id)) + rows = list((await session.exec(stmt)).all()) + if not is_superuser and cross_user: + rows = await filter_visible_resources( + user, + resource_type="connection", + candidates=rows, + owner_extractor=lambda item: user.id + if item.ownership_mode == ConnectionOwnershipMode.INSTANCE.value + else item.owner_id, + act="read", + ) + secret_ids = ( + set( + ( + await session.exec( + select(ConnectionSecret.connection_id).where( + col(ConnectionSecret.connection_id).in_([row.id for row in rows]) + ) + ) + ).all() + ) + if rows + else set() + ) + return [self.to_read(row, has_credentials=row.id in secret_ids) for row in rows] + + async def get_for_user( + self, + session: AsyncSession, + *, + user: User | UserRead, + connection_id: UUID, + for_update: bool = False, + ) -> Connection | None: + authz = get_authorization_service() + may_fetch_cross_user = bool(getattr(user, "is_superuser", False)) or ( + await authz.is_enabled() and await authz.supports_cross_user_fetch() + ) + stmt = select(Connection).where(Connection.id == connection_id) + if not may_fetch_cross_user: + stmt = stmt.where( + or_( + Connection.owner_id == user.id, + Connection.ownership_mode == ConnectionOwnershipMode.INSTANCE.value, + ) + ) + if for_update: + stmt = stmt.with_for_update().execution_options(populate_existing=True) + return (await session.exec(stmt)).first() + + async def has_credentials(self, session: AsyncSession, connection_id: UUID) -> bool: + return await session.get(ConnectionSecret, connection_id) is not None + + async def revoke(self, session: AsyncSession, row: Connection) -> ConnectionRead: + secret = await session.get(ConnectionSecret, row.id) + if secret is not None: + await session.delete(secret) + row.status = PersistedConnectionStatus.REVOKED.value + row.health = ConnectionHealth.UNHEALTHY.value + row.health_checked_at = _utc_now() + row.updated_at = row.health_checked_at + session.add(row) + await session.flush() + await session.refresh(row) + return self.to_read(row, has_credentials=False) + + async def delete(self, session: AsyncSession, row: Connection) -> None: + await session.delete(row) + await session.flush() + + async def check_health( + self, + session: AsyncSession, + *, + row: Connection, + principal: ExecutionPrincipal, + required_scopes: frozenset[str] = frozenset(), + ) -> ConnectionRead: + try: + await self._resolved_from_row(session, row=row, principal=principal, required_scopes=required_scopes) + except AuthExpiredError: + row.status = PersistedConnectionStatus.EXPIRED.value + row.health = ConnectionHealth.UNHEALTHY.value + except IntegrationError: + row.health = ConnectionHealth.UNHEALTHY.value + else: + row.status = PersistedConnectionStatus.READY.value + row.health = ConnectionHealth.HEALTHY.value + row.health_checked_at = _utc_now() + row.updated_at = row.health_checked_at + session.add(row) + await session.flush() + await session.refresh(row) + return self.to_read(row, has_credentials=await self.has_credentials(session, row.id)) + + async def resolve(self, request: ConnectionResolutionRequest) -> ResolvedCredential: + async with session_scope() as session: + candidates = list( + ( + await session.exec( + select(Connection) + .where( + Connection.provider_key == request.ref.provider, + Connection.name == request.ref.name, + ) + .order_by(col(Connection.id)) + ) + ).all() + ) + row = await self._select_authorized_candidate(candidates, request) + if row is None: + raise ConnectionUnresolvedError(request.ref.to_handle(), provider=request.ref.provider) + return await self._resolved_from_row( + session, + row=row, + principal=request.principal, + required_scopes=request.required_scopes, + ) + + async def describe(self, ref: ConnectionRef, principal: ExecutionPrincipal) -> ConnectionStatus | None: + request = ConnectionResolutionRequest(ref=ref, principal=principal) + try: + credential = await self.resolve(request) + except AuthExpiredError: + return ConnectionStatus(ref=ref, status="expired") + except ScopeMissingError: + return ConnectionStatus(ref=ref, status="scope_missing") + except ConnectionUnresolvedError: + return ConnectionStatus(ref=ref, status="missing") + except IntegrationError: + return ConnectionStatus(ref=ref, status="unavailable") + return ConnectionStatus( + ref=ref, + status="ready", + granted_scopes=credential.granted_scopes, + account=credential.account, + ) + + async def _select_authorized_candidate( + self, + candidates: list[Connection], + request: ConnectionResolutionRequest, + ) -> Connection | None: + own: list[Connection] = [] + instance: list[Connection] = [] + shared: list[Connection] = [] + for row in candidates: + if row.ownership_mode == ConnectionOwnershipMode.INSTANCE.value: + instance.append(row) + elif request.principal.user_id is not None and str(row.owner_id) == str(request.principal.user_id): + own.append(row) + else: + shared.append(row) + # An owned record shadows the instance fallback. If its execution + # policy denies this principal, do not silently switch identities by + # resolving an instance credential with the same handle. + for group in (own, instance): + if not group: + continue + row = group[0] + authorization_error = self.authorize_principal( + request, + connection_owner_id=str(row.owner_id) if row.owner_id is not None else None, + owner_kind=row.ownership_mode, + allow_non_interactive=row.allow_non_interactive, + ) + if authorization_error is not None: + raise authorization_error + return row + if request.principal.user_id is None or not shared: + return None + shared = [row for row in shared if request.principal.interactive or row.allow_non_interactive] + if not shared: + return None + settings = get_settings_service() + authz = get_authorization_service() + if not settings.auth_settings.AUTHZ_ENABLED or not await authz.supports_cross_user_fetch(): + return None + try: + user_id = UUID(str(request.principal.user_id)) + except ValueError: + return None + decisions = await authz.batch_enforce( + user_id=user_id, + domain="*", + requests=[(f"connection:{row.id}", "execute") for row in shared], + context={"execution_principal_kind": request.principal.kind}, + ) + authorized = [row for row, allowed in zip(shared, decisions, strict=True) if allowed] + # A handle is intentionally owner-neutral. More than one shared match is + # ambiguous, so fail closed instead of selecting credential material by + # incidental database order. + return authorized[0] if len(authorized) == 1 else None + + async def _resolved_from_row( + self, + session: AsyncSession, + *, + row: Connection, + principal: ExecutionPrincipal, + required_scopes: frozenset[str], + ) -> ResolvedCredential: + request = ConnectionResolutionRequest( + ref=ConnectionRef(provider=row.provider_key, name=row.name), + principal=principal, + required_scopes=required_scopes, + ) + portable_error = self.authorize_principal( + request, + connection_owner_id=str(row.owner_id) if row.owner_id is not None else None, + owner_kind=row.ownership_mode, + allow_non_interactive=row.allow_non_interactive, + ) + if portable_error is not None: + is_explicit_share = ( + row.ownership_mode == ConnectionOwnershipMode.USER.value + and principal.user_id is not None + and str(row.owner_id) != str(principal.user_id) + ) + # The portable floor deliberately reports an owner mismatch for a + # shared connection. Callers reach this private method only after + # the host authorization service has approved that share. Every + # other portable denial, including anonymous and non-interactive + # use without the per-connection opt-in, remains authoritative. + if not is_explicit_share or (not principal.interactive and not row.allow_non_interactive): + raise portable_error + if row.status == PersistedConnectionStatus.REVOKED.value: + raise ConnectionUnresolvedError(request.ref.to_handle(), provider=row.provider_key) + secret = await session.get(ConnectionSecret, row.id) + if secret is None: + raise ConnectionUnresolvedError(request.ref.to_handle(), provider=row.provider_key) + try: + payload = _decrypt_credential_payload(secret.encrypted_payload) + except ConnectionSecretError as exc: + raise ConnectionUnresolvedError(request.ref.to_handle(), provider=row.provider_key) from exc + expires_at = _parse_expiry(payload.get("expires_at")) + if expires_at is not None and expires_at <= _utc_now(): + raise AuthExpiredError(provider=row.provider_key) + granted = frozenset(row.granted_scopes) + missing = required_scopes - granted + if missing: + raise ScopeMissingError(missing, provider=row.provider_key) + identity = ExecutingIdentityDescriptor.model_validate(row.executing_identity) + return ResolvedCredential( + access_token=SecretStr(payload["access_token"]), + token_type=str(payload.get("token_type") or "Bearer"), + expires_at=expires_at, + granted_scopes=granted, + scopes_verified=True, + account=identity.account, + connection_id=str(row.id), + owner_kind=row.ownership_mode, + provider=row.provider_key, + name=row.name, + ) + + @staticmethod + def to_read(row: Connection, *, has_credentials: bool) -> ConnectionRead: + return ConnectionRead.model_validate( + { + **row.model_dump(), + "has_credentials": has_credentials, + } + ) diff --git a/src/backend/base/langflow/services/database/models/__init__.py b/src/backend/base/langflow/services/database/models/__init__.py index 9ffd20afe0e4..72099b413f8e 100644 --- a/src/backend/base/langflow/services/database/models/__init__.py +++ b/src/backend/base/langflow/services/database/models/__init__.py @@ -22,6 +22,7 @@ is_sso_client_secret_envelope, ) from .catalog_policy import CatalogPolicyMode, CatalogPolicyRule, CatalogPolicyScope, CatalogResourceKind +from .connection import Connection, ConnectionSecret from .deployment import Deployment from .deployment_provider_account import DeploymentProviderAccount from .file import File @@ -59,6 +60,8 @@ "CatalogPolicyRule", "CatalogPolicyScope", "CatalogResourceKind", + "Connection", + "ConnectionSecret", "Deployment", "DeploymentProviderAccount", "ExecutionSignal", diff --git a/src/backend/base/langflow/services/database/models/connection/__init__.py b/src/backend/base/langflow/services/database/models/connection/__init__.py new file mode 100644 index 000000000000..3c60c24ef52a --- /dev/null +++ b/src/backend/base/langflow/services/database/models/connection/__init__.py @@ -0,0 +1,25 @@ +from .model import Connection, ConnectionBase, ConnectionSecret +from .schemas import ( + ConnectionCreate, + ConnectionCredentialWrite, + ConnectionHealth, + ConnectionOwnershipMode, + ConnectionRead, + ConnectionTestRequest, + ExecutingIdentityDescriptor, + PersistedConnectionStatus, +) + +__all__ = [ + "Connection", + "ConnectionBase", + "ConnectionCreate", + "ConnectionCredentialWrite", + "ConnectionHealth", + "ConnectionOwnershipMode", + "ConnectionRead", + "ConnectionSecret", + "ConnectionTestRequest", + "ExecutingIdentityDescriptor", + "PersistedConnectionStatus", +] diff --git a/src/backend/base/langflow/services/database/models/connection/model.py b/src/backend/base/langflow/services/database/models/connection/model.py new file mode 100644 index 000000000000..8cc4d24efa46 --- /dev/null +++ b/src/backend/base/langflow/services/database/models/connection/model.py @@ -0,0 +1,100 @@ +"""Database tables for connection metadata and encrypted token material.""" + +from __future__ import annotations + +from datetime import datetime # noqa: TC003 - SQLModel resolves annotations at runtime +from uuid import UUID, uuid4 + +import sqlalchemy as sa +from sqlalchemy import CheckConstraint, ForeignKey, Index +from sqlmodel import JSON, Column, DateTime, Field, SQLModel, func + +from langflow.schema.serialize import UUIDstr # noqa: TC001 - SQLModel resolves annotations at runtime +from langflow.services.database.models.connection.schemas import ( + ConnectionHealth, + ConnectionOwnershipMode, + PersistedConnectionStatus, +) + + +class ConnectionBase(SQLModel): + provider_key: str = Field(max_length=120) + name: str = Field(max_length=64) + display_name: str = Field(max_length=255) + ownership_mode: str = Field(default=ConnectionOwnershipMode.USER.value, max_length=16) + status: str = Field(default=PersistedConnectionStatus.PENDING.value, max_length=16) + health: str = Field(default=ConnectionHealth.UNKNOWN.value, max_length=16) + granted_scopes: list[str] = Field(default_factory=list, sa_column=Column(JSON, nullable=False)) + executing_identity: dict = Field(default_factory=dict, sa_column=Column(JSON, nullable=False)) + allow_non_interactive: bool = Field(default=False, nullable=False) + + +class Connection(ConnectionBase, table=True): # type: ignore[call-arg] + """Non-secret connection metadata.""" + + __tablename__ = "connection" + __table_args__ = ( + CheckConstraint( + "(ownership_mode = 'user' AND owner_id IS NOT NULL) OR (ownership_mode = 'instance' AND owner_id IS NULL)", + name="ck_connection_owner_mode", + ), + CheckConstraint( + "status IN ('pending', 'ready', 'expired', 'revoked', 'error')", + name="ck_connection_status", + ), + CheckConstraint( + "health IN ('unknown', 'healthy', 'unhealthy')", + name="ck_connection_health", + ), + Index( + "uq_connection_user_provider_name", + "owner_id", + "provider_key", + "name", + unique=True, + sqlite_where=sa.text("ownership_mode = 'user'"), + postgresql_where=sa.text("ownership_mode = 'user'"), + ), + Index( + "uq_connection_instance_provider_name", + "provider_key", + "name", + unique=True, + sqlite_where=sa.text("ownership_mode = 'instance'"), + postgresql_where=sa.text("ownership_mode = 'instance'"), + ), + ) + + id: UUID = Field(default_factory=uuid4, primary_key=True) + owner_id: UUIDstr | None = Field( + default=None, + sa_column=Column(sa.Uuid(), ForeignKey("user.id", ondelete="CASCADE"), nullable=True, index=True), + ) + health_checked_at: datetime | None = Field(default=None, sa_column=Column(DateTime(timezone=True), nullable=True)) + created_at: datetime | None = Field( + default=None, + sa_column=Column(DateTime(timezone=True), server_default=func.now(), nullable=False), + ) + updated_at: datetime | None = Field( + default=None, + sa_column=Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now(), nullable=False), + ) + + +class ConnectionSecret(SQLModel, table=True): # type: ignore[call-arg] + """Encrypted connection credential envelope, isolated from metadata reads.""" + + __tablename__ = "connection_secret" + + connection_id: UUID = Field( + sa_column=Column(sa.Uuid(), ForeignKey("connection.id", ondelete="CASCADE"), nullable=False, primary_key=True), + ) + encrypted_payload: str = Field(sa_column=Column(sa.Text(), nullable=False)) + created_at: datetime | None = Field( + default=None, + sa_column=Column(DateTime(timezone=True), server_default=func.now(), nullable=False), + ) + updated_at: datetime | None = Field( + default=None, + sa_column=Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now(), nullable=False), + ) diff --git a/src/backend/base/langflow/services/database/models/connection/schemas.py b/src/backend/base/langflow/services/database/models/connection/schemas.py new file mode 100644 index 000000000000..cb541eb0fe90 --- /dev/null +++ b/src/backend/base/langflow/services/database/models/connection/schemas.py @@ -0,0 +1,123 @@ +"""Public and internal schemas for persisted integration connections.""" + +from __future__ import annotations + +from datetime import datetime +from enum import Enum +from uuid import UUID + +from lfx.integrations.capabilities import IntegrationIdentity +from lfx.integrations.models import CONNECTION_NAME_PATTERN, PROVIDER_ID_PATTERN, ConnectionAccount +from pydantic import BaseModel, ConfigDict, Field, SecretStr, StrictStr, field_validator + + +class ConnectionOwnershipMode(str, Enum): + USER = "user" + INSTANCE = "instance" + + +class PersistedConnectionStatus(str, Enum): + PENDING = "pending" + READY = "ready" + EXPIRED = "expired" + REVOKED = "revoked" + ERROR = "error" + + +class ConnectionHealth(str, Enum): + UNKNOWN = "unknown" + HEALTHY = "healthy" + UNHEALTHY = "unhealthy" + + +class ExecutingIdentityDescriptor(BaseModel): + """Non-secret identity metadata shown by connection pickers.""" + + model_config = ConfigDict(extra="forbid") + + identity: IntegrationIdentity + account: ConnectionAccount | None = None + + +class ConnectionCredentialWrite(BaseModel): + """Write-only credential material encrypted before persistence.""" + + model_config = ConfigDict(extra="forbid") + + access_token: SecretStr = Field(min_length=1) + refresh_token: SecretStr | None = None + token_type: StrictStr = Field(default="Bearer", min_length=1, max_length=32) + expires_at: datetime | None = None + + +class ConnectionCreate(BaseModel): + """Create metadata plus optional direct-provisioned credential material.""" + + model_config = ConfigDict(extra="forbid") + + provider_key: StrictStr = Field(pattern=PROVIDER_ID_PATTERN, max_length=120) + name: StrictStr = Field(pattern=CONNECTION_NAME_PATTERN, max_length=64) + display_name: StrictStr = Field(min_length=1, max_length=255) + ownership_mode: ConnectionOwnershipMode = ConnectionOwnershipMode.USER + granted_scopes: list[StrictStr] = Field(default_factory=list, max_length=512) + executing_identity: ExecutingIdentityDescriptor + allow_non_interactive: bool = False + credentials: ConnectionCredentialWrite | None = None + + @field_validator("display_name") + @classmethod + def _display_name_not_blank(cls, value: str) -> str: + value = value.strip() + if not value: + msg = "display_name must not be blank" + raise ValueError(msg) + return value + + @field_validator("granted_scopes") + @classmethod + def _scopes_are_unique_and_nonblank(cls, value: list[str]) -> list[str]: + normalized = [scope.strip() for scope in value] + if any(not scope for scope in normalized): + msg = "granted_scopes must not contain blank values" + raise ValueError(msg) + if len(set(normalized)) != len(normalized): + msg = "granted_scopes must not contain duplicates" + raise ValueError(msg) + return normalized + + +class ConnectionRead(BaseModel): + """Credential-free connection metadata returned by every API route.""" + + model_config = ConfigDict(from_attributes=True) + + id: UUID + owner_id: UUID | None + ownership_mode: ConnectionOwnershipMode + provider_key: str + name: str + display_name: str + status: PersistedConnectionStatus + health: ConnectionHealth + granted_scopes: list[str] + executing_identity: ExecutingIdentityDescriptor + allow_non_interactive: bool + has_credentials: bool + health_checked_at: datetime | None + created_at: datetime + updated_at: datetime + + +class ConnectionTestRequest(BaseModel): + model_config = ConfigDict(extra="forbid") + + required_scopes: list[StrictStr] = Field(default_factory=list, max_length=512) + + @field_validator("required_scopes") + @classmethod + def _required_scopes_are_unique_and_nonblank(cls, value: list[str]) -> list[str]: + normalized = [scope.strip() for scope in value] + if any(not scope for scope in normalized) or len(set(normalized)) != len(normalized): + msg = "required_scopes must contain unique, non-blank values" + raise ValueError(msg) + return normalized diff --git a/src/backend/base/langflow/services/deployment_artifacts/__init__.py b/src/backend/base/langflow/services/deployment_artifacts/__init__.py index 878b06fea048..228d4ed1e6af 100644 --- a/src/backend/base/langflow/services/deployment_artifacts/__init__.py +++ b/src/backend/base/langflow/services/deployment_artifacts/__init__.py @@ -9,6 +9,7 @@ ProjectArtifactLimitError, ProjectArtifactLimits, ProjectArtifactNotFoundError, + ProjectArtifactRequiredConnection, build_project_artifact, ) @@ -21,5 +22,6 @@ "ProjectArtifactLimitError", "ProjectArtifactLimits", "ProjectArtifactNotFoundError", + "ProjectArtifactRequiredConnection", "build_project_artifact", ] diff --git a/src/backend/base/langflow/services/deployment_artifacts/builder.py b/src/backend/base/langflow/services/deployment_artifacts/builder.py index 765db0ed795c..5d388e8c127b 100644 --- a/src/backend/base/langflow/services/deployment_artifacts/builder.py +++ b/src/backend/base/langflow/services/deployment_artifacts/builder.py @@ -13,6 +13,7 @@ from functools import partial from typing import TYPE_CHECKING, Any, TypeVar +from lfx.integrations.models import ConnectionRef from sqlmodel import col, select from langflow.services.authorization import ( @@ -102,6 +103,16 @@ class ProjectArtifactFlow: sha256: str size: int required_variables: tuple[str, ...] + required_connections: tuple[ProjectArtifactRequiredConnection, ...] + + +@dataclass(frozen=True, order=True, slots=True) +class ProjectArtifactRequiredConnection: + """One non-secret connection handle and its static scope requirements.""" + + provider: str + name: str + scopes: tuple[str, ...] = () @dataclass(frozen=True, slots=True) @@ -154,7 +165,65 @@ def _zip_info(path: str) -> zipfile.ZipInfo: return info -def _normalized_flow_bytes(snapshot: _FlowSnapshot) -> tuple[bytes, tuple[str, ...]]: +def _collect_required_connections(flow_data: object) -> tuple[ProjectArtifactRequiredConnection, ...]: + """Collect connection refs from regular and grouped nodes without recursion.""" + if not isinstance(flow_data, dict): + return () + nodes = flow_data.get("nodes") + if not isinstance(nodes, list): + return () + collected: dict[tuple[str, str], set[str]] = {} + node_frames = [iter(nodes)] + while node_frames: + try: + node = next(node_frames[-1]) + except StopIteration: + node_frames.pop() + continue + if not isinstance(node, dict): + continue + node_inner = node.get("data", {}).get("node") if isinstance(node.get("data"), dict) else None + if not isinstance(node_inner, dict): + continue + template = node_inner.get("template") + if isinstance(template, dict): + for field_value in template.values(): + if not isinstance(field_value, dict) or field_value.get("type") != "connection_ref": + continue + value = field_value.get("value") + if value in (None, ""): + continue + try: + ref = ConnectionRef.parse(value) + except ValueError as exc: + msg = "project artifact contains an invalid connection reference" + raise ProjectArtifactError(msg) from exc + declared_provider = field_value.get("provider") + if not isinstance(declared_provider, str) or declared_provider != ref.provider: + msg = "project artifact connection reference does not match its declared provider" + raise ProjectArtifactError(msg) + raw_scopes = field_value.get("required_scopes", []) + if not isinstance(raw_scopes, list) or any( + not isinstance(scope, str) or not scope.strip() for scope in raw_scopes + ): + msg = f"connection {ref.to_handle()!r} has invalid required scopes" + raise ProjectArtifactError(msg) + collected.setdefault((ref.provider, ref.name), set()).update(scope.strip() for scope in raw_scopes) + nested_flow = node_inner.get("flow") + if isinstance(nested_flow, dict): + nested_data = nested_flow.get("data") + nested_nodes = nested_data.get("nodes") if isinstance(nested_data, dict) else None + if isinstance(nested_nodes, list): + node_frames.append(iter(nested_nodes)) + return tuple( + ProjectArtifactRequiredConnection(provider=provider, name=name, scopes=tuple(sorted(scopes))) + for (provider, name), scopes in sorted(collected.items()) + ) + + +def _normalized_flow_bytes( + snapshot: _FlowSnapshot, +) -> tuple[bytes, tuple[str, ...], tuple[ProjectArtifactRequiredConnection, ...]]: # Scrubbing and volatile-field removal mutate nested values in place. Copy # first so aliases held by the snapshot or persisted Flow data stay intact. scrubbed = deepcopy(snapshot.payload) @@ -162,6 +231,7 @@ def _normalized_flow_bytes(snapshot: _FlowSnapshot) -> tuple[bytes, tuple[str, . # the serving side can provision credentials under the same names; the # collected names feed the manifest's required-variables listing. variable_references: set[str] = set() + required_connections = _collect_required_connections(scrubbed.get("data")) scrubbed["data"] = strip_secret_field_values_in_place(scrubbed.get("data"), variable_references=variable_references) # Deployment packages retain runtime-native code strings. The normal git # export path splits code into one list element per line, which is useful @@ -177,7 +247,7 @@ def _normalized_flow_bytes(snapshot: _FlowSnapshot) -> tuple[bytes, tuple[str, . if isinstance(node, dict): for key in _VOLATILE_NODE_FIELDS: node.pop(key, None) - return _canonical_json_bytes(scrubbed), tuple(sorted(variable_references)) + return _canonical_json_bytes(scrubbed), tuple(sorted(variable_references)), required_connections def _json_string_size(value: str) -> int: @@ -317,7 +387,7 @@ def _build_archive( for snapshot in snapshots: path = f"flows/{snapshot.flow_id}.json" - content, required_variables = _normalized_flow_bytes(snapshot) + content, required_variables, required_connections = _normalized_flow_bytes(snapshot) size = len(content) if size > limits.max_flow_bytes: msg = f"flow file {snapshot.flow_id} is {size} bytes, exceeding the {limits.max_flow_bytes}-byte limit" @@ -335,14 +405,25 @@ def _build_archive( sha256=hashlib.sha256(content).hexdigest(), size=size, required_variables=required_variables, + required_connections=required_connections, ) ) - manifest = { - # v2 is already assigned to flows[].version_id. Dependencies therefore - # use v3 so an older reader refuses them instead of deploying without - # provisioning resources the packaged flows require. - "schema_version": 3 if dependencies else 1, + required_connections_by_handle: dict[tuple[str, str], set[str]] = {} + for flow in flow_entries: + for connection in flow.required_connections: + required_connections_by_handle.setdefault((connection.provider, connection.name), set()).update( + connection.scopes + ) + manifest_required_connections = [ + {"provider": provider, "name": name, "scopes": sorted(scopes)} + for (provider, name), scopes in sorted(required_connections_by_handle.items()) + ] + manifest: dict[str, Any] = { + # v2 is already assigned to flows[].version_id. Dependencies use v3, + # and connection requirements use v4, so an older reader refuses an + # artifact instead of deploying without provisioning required resources. + "schema_version": 4 if manifest_required_connections else (3 if dependencies else 1), "project": {"id": str(project_id), "name": project_name}, # Names of every load_from_db-bound global variable the packaged flows # reference; the deploy target must provision each name before serving. @@ -355,10 +436,22 @@ def _build_archive( "sha256": flow.sha256, "size": flow.size, "required_variables": list(flow.required_variables), + **( + { + "required_connections": [ + {"provider": item.provider, "name": item.name, "scopes": list(item.scopes)} + for item in flow.required_connections + ] + } + if flow.required_connections + else {} + ), } for flow in flow_entries ], } + if manifest_required_connections: + manifest["required_connections"] = manifest_required_connections if dependencies: manifest["dependencies"] = dependencies manifest_bytes = _canonical_json_bytes(manifest) diff --git a/src/backend/base/langflow/services/deps.py b/src/backend/base/langflow/services/deps.py index de57739ab357..1d20e900b20c 100644 --- a/src/backend/base/langflow/services/deps.py +++ b/src/backend/base/langflow/services/deps.py @@ -27,6 +27,7 @@ from lfx.services.auth.base import BaseAuthService # noqa: TC002 from lfx.services.authorization.base import BaseAuthorizationService # noqa: TC002 from lfx.services.catalog_policy.base import BaseCatalogPolicyService # noqa: TC002 +from lfx.services.connection.base import BaseConnectionResolverService # noqa: TC002 from lfx.services.policy_bundle.base import BasePolicyBundleService # noqa: TC002 from lfx.services.settings.service import SettingsService # noqa: TC002 @@ -300,6 +301,13 @@ def get_authorization_service() -> BaseAuthorizationService: return get_service(ServiceType.AUTHORIZATION_SERVICE, AuthorizationServiceFactory()) +def get_connection_resolver_service() -> BaseConnectionResolverService: + """Retrieve Langflow's active host-pluggable connection resolver.""" + from langflow.services.connection.factory import ConnectionResolverServiceFactory + + return get_service(ServiceType.CONNECTION_RESOLVER_SERVICE, ConnectionResolverServiceFactory()) + + def get_catalog_policy_service() -> BaseCatalogPolicyService: """Retrieve catalog policy through LFX's validated fail-open dependency.""" from lfx.services.deps import get_catalog_policy_service as get_lfx_catalog_policy_service diff --git a/src/backend/base/langflow/services/factory.py b/src/backend/base/langflow/services/factory.py index 564b66a4c32a..bc2221ab37ff 100644 --- a/src/backend/base/langflow/services/factory.py +++ b/src/backend/base/langflow/services/factory.py @@ -79,7 +79,7 @@ def import_all_services_into_a_dict(): # Shared services live in lfx so both the standalone runtime and # the Langflow application use the same contract and instance. if service_name == "connection_resolver": - module_name = "lfx.services.connection.env_resolver" + module_name = "langflow.services.connection.service" elif service_name in {"mcp_composer", "model_provider_policy", "policy_bundle"}: module_name = f"lfx.services.{service_name}.service" else: diff --git a/src/backend/base/langflow/services/utils.py b/src/backend/base/langflow/services/utils.py index 045bbec211d7..53c3a6683f4b 100644 --- a/src/backend/base/langflow/services/utils.py +++ b/src/backend/base/langflow/services/utils.py @@ -580,6 +580,8 @@ def register_all_service_factories() -> None: from langflow.services.catalog_policy.service import LangflowCatalogPolicyService from langflow.services.chat import factory as chat_factory from langflow.services.checkpoint import factory as checkpoint_factory + from langflow.services.connection import factory as connection_factory + from langflow.services.connection.service import DatabaseConnectionResolverService from langflow.services.database import factory as database_factory from langflow.services.job_queue import factory as job_queue_factory from langflow.services.session import factory as session_factory @@ -612,6 +614,15 @@ def register_all_service_factories() -> None: service_manager.register_factory(task_factory.TaskServiceFactory()) service_manager.register_factory(store_factory.StoreServiceFactory()) service_manager.register_factory(shared_component_cache_factory.SharedComponentCacheServiceFactory()) + # Standalone LFX keeps its environment resolver. Full Langflow resolves + # handles from encrypted connection rows unless a host plugin overrides + # this service through lfx.toml. + service_manager.register_service_class( + ServiceType.CONNECTION_RESOLVER_SERVICE, + DatabaseConnectionResolverService, + override=True, + ) + service_manager.register_factory(connection_factory.ConnectionResolverServiceFactory()) # Override LFX's no-op auth service with Langflow's full JWT implementation service_manager.register_service_class(ServiceType.AUTH_SERVICE, AuthService, override=True) service_manager.register_factory(auth_factory.AuthServiceFactory()) diff --git a/src/backend/base/langflow/utils/flow_secrets.py b/src/backend/base/langflow/utils/flow_secrets.py index 12f2a5ea6c7b..c5ab2f125124 100644 --- a/src/backend/base/langflow/utils/flow_secrets.py +++ b/src/backend/base/langflow/utils/flow_secrets.py @@ -290,6 +290,22 @@ def _strip_table_rows_in_place(field: dict, reference_columns: frozenset[str], v def _strip_template_field_value(field: dict, variable_references: set[str] | None = None) -> None: """Strip a template field according to metadata and value shape.""" + field_type = str(field.get("type") or "").lower() + if field_type == "connection_ref": + # Connection handles are non-secret deployment references. Validate the + # shape before preserving one so a malformed legacy value does not get + # mistaken for a deployable connection requirement. + value = field.get("value") + if value in (None, ""): + return + try: + from lfx.integrations.models import ConnectionRef + + field["value"] = ConnectionRef.parse(value).to_handle() + except (TypeError, ValueError): + field["value"] = None + return + if ( variable_references is not None and field.get("load_from_db") @@ -309,7 +325,6 @@ def _strip_template_field_value(field: dict, variable_references: set[str] | Non field["value"] = None return - field_type = str(field.get("type") or "").lower() input_type = str(field.get("_input_type") or "").lower() if field_type == "mcp" or input_type == "mcpinput": value = field.get("value") diff --git a/src/backend/tests/unit/alembic/test_connection_migration.py b/src/backend/tests/unit/alembic/test_connection_migration.py new file mode 100644 index 000000000000..c168462e134d --- /dev/null +++ b/src/backend/tests/unit/alembic/test_connection_migration.py @@ -0,0 +1,35 @@ +"""Portable migration coverage for connection metadata and secret isolation.""" + +from alembic import command +from sqlalchemy import create_engine, inspect + +from .test_migration_execution import _engine_url, _make_alembic_cfg, db_url # noqa: F401 + +_PRIOR_REVISION = "c9f2e5a7b1d4" # pragma: allowlist secret +_REVISION = "f3b6a9d2e4c1" # pragma: allowlist secret + + +def test_connection_migration_round_trip_sqlite_and_postgres(db_url): # noqa: F811 + config = _make_alembic_cfg(db_url) + command.upgrade(config, _PRIOR_REVISION) + command.upgrade(config, _REVISION) + + engine = create_engine(_engine_url(db_url)) + try: + with engine.connect() as connection: + inspector = inspect(connection) + assert {"connection", "connection_secret"} <= set(inspector.get_table_names()) + assert "encrypted_payload" not in {column["name"] for column in inspector.get_columns("connection")} + assert "encrypted_payload" in {column["name"] for column in inspector.get_columns("connection_secret")} + finally: + engine.dispose() + + command.downgrade(config, _PRIOR_REVISION) + engine = create_engine(_engine_url(db_url)) + try: + with engine.connect() as connection: + tables = set(inspect(connection).get_table_names()) + assert "connection" not in tables + assert "connection_secret" not in tables + finally: + engine.dispose() diff --git a/src/backend/tests/unit/api/v1/test_authz_admin_routes.py b/src/backend/tests/unit/api/v1/test_authz_admin_routes.py index 5846759fa79a..0ff5a87b1672 100644 --- a/src/backend/tests/unit/api/v1/test_authz_admin_routes.py +++ b/src/backend/tests/unit/api/v1/test_authz_admin_routes.py @@ -305,6 +305,7 @@ def test_role_create_rejects_other_nested_component_permissions(permission): "share:write", # write isn't a share action "variable:execute", # variables aren't executed "voice:execute", # websocket execution is governed by flow:execute + "connection:ingest", # ingest is knowledge_base-only ], ) def test_role_create_rejects_non_canonical_permission_slugs(bad_slug): @@ -333,6 +334,7 @@ def test_role_create_rejects_non_canonical_permission_slugs(bad_slug): "variable:write", "project:delete", "voice:read", + "connection:execute", # Wildcard remains valid on every resource. "flow:*", "share:*", diff --git a/src/backend/tests/unit/api/v1/test_authz_share_routes.py b/src/backend/tests/unit/api/v1/test_authz_share_routes.py index 7234d4729f50..8fc186b55149 100644 --- a/src/backend/tests/unit/api/v1/test_authz_share_routes.py +++ b/src/backend/tests/unit/api/v1/test_authz_share_routes.py @@ -426,6 +426,31 @@ async def test_create_share_keeps_public_read_for_non_flow_resources(patch_authz assert session.committed == 1 +@pytest.mark.asyncio +async def test_create_share_rejects_public_connection(patch_authz, silence_audit): # noqa: ARG001 + from langflow.services.database.models.connection import Connection + + patch_authz(cross_user=False, enabled=False) + + owner = _make_user() + connection = SimpleNamespace(id=uuid4(), owner_id=owner.id) + session = _FakeAsyncSession({(Connection, connection.id): connection}) + payload = ShareCreate( + resource_type="connection", + resource_id=connection.id, + scope=ShareScope.PUBLIC.value, + permission_level=SharePermissionLevel.EXECUTE.value, + ) + + with pytest.raises(HTTPException) as excinfo: + await shares_module.create_share(payload=payload, current_user=owner, session=session) + + assert excinfo.value.status_code == 422 + assert excinfo.value.detail == "Connections cannot be shared publicly." + assert session.added == [] + assert session.flushed == 0 + + # --------------------------------------------------------------------------- # # PATCH — same floor # --------------------------------------------------------------------------- # diff --git a/src/backend/tests/unit/api/v1/test_connections.py b/src/backend/tests/unit/api/v1/test_connections.py new file mode 100644 index 000000000000..43e6e5cbd53d --- /dev/null +++ b/src/backend/tests/unit/api/v1/test_connections.py @@ -0,0 +1,212 @@ +"""API coverage for connection ownership and secret-safe serialization.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING +from uuid import UUID, uuid4 + +import pytest +from langflow.services.auth.utils import get_auth_service +from langflow.services.database.models.connection import ConnectionSecret +from langflow.services.database.models.user.model import User +from langflow.services.deps import get_connection_resolver_service, session_scope +from lfx.integrations.errors import ConnectionNotAuthorizedError, ScopeMissingError +from lfx.integrations.models import ConnectionRef, ConnectionResolutionRequest +from lfx.services.authorization.base import ExecutionPrincipal +from sqlmodel import select + +if TYPE_CHECKING: + from httpx import AsyncClient + +pytestmark = pytest.mark.no_blockbuster + + +def _payload( + *, + ownership_mode: str = "user", + name: str = "work", + allow_non_interactive: bool = False, +) -> dict: + return { + "provider_key": "google_workspace", + "name": name, + "display_name": "Work Google", + "ownership_mode": ownership_mode, + "granted_scopes": ["calendar.readonly"], + "executing_identity": { + "identity": "user_delegated", + "account": {"id": "account-123", "display": "Work", "tenant_id": "tenant-123"}, + }, + "allow_non_interactive": allow_non_interactive, + "credentials": { + "access_token": "access-token-do-not-return", + "refresh_token": "refresh-token-do-not-return", + "token_type": "Bearer", + }, + } + + +def _assert_no_credentials(value: object) -> None: + forbidden_keys = {"access_token", "refresh_token", "encrypted_payload", "credentials"} + if isinstance(value, dict): + assert forbidden_keys.isdisjoint(value) + for item in value.values(): + _assert_no_credentials(item) + elif isinstance(value, list): + for item in value: + _assert_no_credentials(item) + elif isinstance(value, str): + assert value not in {"access-token-do-not-return", "refresh-token-do-not-return"} + + +@pytest.mark.usefixtures("active_user") +async def test_connection_responses_never_include_tokens( + client: AsyncClient, + logged_in_headers: dict[str, str], +) -> None: + created = await client.post("api/v1/connections", json=_payload(), headers=logged_in_headers) + assert created.status_code == 201, created.text + body = created.json() + assert body["has_credentials"] is True + assert body["status"] == "ready" + _assert_no_credentials(body) + + listed = await client.get("api/v1/connections", headers=logged_in_headers) + assert listed.status_code == 200, listed.text + _assert_no_credentials(listed.json()) + + filtered = await client.get("api/v1/connections?provider=google_workspace", headers=logged_in_headers) + assert filtered.status_code == 200, filtered.text + assert [item["id"] for item in filtered.json()] == [body["id"]] + + no_matches = await client.get("api/v1/connections?provider=slack", headers=logged_in_headers) + assert no_matches.status_code == 200, no_matches.text + assert no_matches.json() == [] + + resolver = get_connection_resolver_service() + ref = ConnectionRef(provider="google_workspace", name="work") + interactive = ExecutionPrincipal( + kind="actor", + user_id=body["owner_id"], + actor_id=body["owner_id"], + interactive=True, + ) + resolved = await resolver.resolve(ConnectionResolutionRequest(ref=ref, principal=interactive)) + assert resolved.access_token.get_secret_value() == "access-token-do-not-return" + assert resolved.granted_scopes == frozenset({"calendar.readonly"}) + + with pytest.raises(ScopeMissingError): + await resolver.resolve( + ConnectionResolutionRequest( + ref=ref, + principal=interactive, + required_scopes=frozenset({"calendar.write"}), + ) + ) + + non_interactive = ExecutionPrincipal( + kind="flow_owner", + user_id=body["owner_id"], + actor_id=body["owner_id"], + interactive=False, + ) + with pytest.raises(ConnectionNotAuthorizedError): + await resolver.resolve(ConnectionResolutionRequest(ref=ref, principal=non_interactive)) + + unattended_created = await client.post( + "api/v1/connections", + json=_payload(name="automation", allow_non_interactive=True), + headers=logged_in_headers, + ) + assert unattended_created.status_code == 201, unattended_created.text + unattended = await resolver.resolve( + ConnectionResolutionRequest( + ref=ConnectionRef(provider="google_workspace", name="automation"), + principal=non_interactive, + ) + ) + assert unattended.access_token.get_secret_value() == "access-token-do-not-return" + + tested = await client.post( + f"api/v1/connections/{body['id']}/test", + json={"required_scopes": ["calendar.readonly"]}, + headers=logged_in_headers, + ) + assert tested.status_code == 200, tested.text + assert tested.json()["health"] == "healthy" + _assert_no_credentials(tested.json()) + + async with session_scope() as session: + stored = ( + await session.exec(select(ConnectionSecret).where(ConnectionSecret.connection_id == UUID(body["id"]))) + ).one() + assert "access-token-do-not-return" not in stored.encrypted_payload + assert "refresh-token-do-not-return" not in stored.encrypted_payload + + revoked = await client.post(f"api/v1/connections/{body['id']}/revoke", headers=logged_in_headers) + assert revoked.status_code == 200, revoked.text + assert revoked.json()["status"] == "revoked" + assert revoked.json()["has_credentials"] is False + _assert_no_credentials(revoked.json()) + + deleted = await client.delete(f"api/v1/connections/{body['id']}", headers=logged_in_headers) + assert deleted.status_code == 204, deleted.text + + +@pytest.mark.usefixtures("active_user") +async def test_non_owner_cannot_test_or_delete_connection( + client: AsyncClient, + logged_in_headers: dict[str, str], +) -> None: + created = await client.post("api/v1/connections", json=_payload(), headers=logged_in_headers) + assert created.status_code == 201, created.text + connection_id = created.json()["id"] + + username = f"other-{uuid4().hex}" + password = "test-non-owner-password" # noqa: S105 # pragma: allowlist secret + async with session_scope() as session: + other = User( + username=username, + password=get_auth_service().get_password_hash(password), + is_active=True, + ) + session.add(other) + await session.flush() + await session.refresh(other) + other_id = other.id + + login = await client.post("api/v1/login", data={"username": username, "password": password}) + assert login.status_code == 200, login.text + headers = {"Authorization": f"Bearer {login.json()['access_token']}"} + tested = await client.post( + f"api/v1/connections/{connection_id}/test", + json={"required_scopes": []}, + headers=headers, + ) + deleted = await client.delete(f"api/v1/connections/{connection_id}", headers=headers) + assert tested.status_code == 404 + assert deleted.status_code == 404 + + async with session_scope() as session: + other = await session.get(User, other_id) + if other is not None: + await session.delete(other) + + +@pytest.mark.usefixtures("active_super_user") +async def test_superuser_can_create_and_list_instance_connection( + client: AsyncClient, + logged_in_headers_super_user: dict[str, str], +) -> None: + created = await client.post( + "api/v1/connections", + json=_payload(ownership_mode="instance"), + headers=logged_in_headers_super_user, + ) + assert created.status_code == 201, created.text + assert created.json()["owner_id"] is None + assert created.json()["ownership_mode"] == "instance" + + listed = await client.get("api/v1/connections", headers=logged_in_headers_super_user) + assert listed.status_code == 200, listed.text + assert [item["id"] for item in listed.json()] == [created.json()["id"]] diff --git a/src/backend/tests/unit/services/authorization/test_guards.py b/src/backend/tests/unit/services/authorization/test_guards.py index ce5b0371063c..1bf3d11f7809 100644 --- a/src/backend/tests/unit/services/authorization/test_guards.py +++ b/src/backend/tests/unit/services/authorization/test_guards.py @@ -18,6 +18,7 @@ set_current_external_access_context, ) from langflow.services.authorization.actions import ( + ConnectionAction, DeploymentAction, FileAction, FlowAction, @@ -36,6 +37,52 @@ install_settings, ) +# ----------------------------------------------------------------------------- # +# ensure_connection_permission +# ----------------------------------------------------------------------------- # + + +@pytest.mark.anyio +async def test_connection_owner_and_scoped_api_key_personas(monkeypatch, fake_user): + """Connection use honors owner override but not a narrower API-key policy.""" + install_settings(monkeypatch, authz_enabled=True) + service = _StubAuthorizationService(allow=False, supports_api_key_scopes=True) + install_authz(monkeypatch, service) + install_audit_recorder(monkeypatch) + connection_id = uuid4() + + await authz_guards.ensure_connection_permission( + fake_user, + ConnectionAction.EXECUTE, + connection_id=connection_id, + connection_owner_id=fake_user.id, + ) + assert service.calls == [] + + set_current_auth_context( + AuthCredentialContext( + method=AUTH_METHOD_API_KEY, + api_key_id=uuid4(), + api_key_source="db", # pragma: allowlist secret + ) + ) + try: + with pytest.raises(HTTPException) as exc_info: + await authz_guards.ensure_connection_permission( + fake_user, + ConnectionAction.EXECUTE, + connection_id=connection_id, + connection_owner_id=fake_user.id, + ) + finally: + clear_current_auth_context() + + assert exc_info.value.status_code == 403 + assert service.calls[-1]["obj"] == f"connection:{connection_id}" + assert service.calls[-1]["act"] == "execute" + assert service.calls[-1]["context"]["connection_owner_id"] == fake_user.id + + # ----------------------------------------------------------------------------- # # ensure_permission # ----------------------------------------------------------------------------- # diff --git a/src/backend/tests/unit/services/deployment_artifacts/test_project_artifact.py b/src/backend/tests/unit/services/deployment_artifacts/test_project_artifact.py index 3d25c4491180..6e23445566d3 100644 --- a/src/backend/tests/unit/services/deployment_artifacts/test_project_artifact.py +++ b/src/backend/tests/unit/services/deployment_artifacts/test_project_artifact.py @@ -152,8 +152,10 @@ async def test_build_project_artifact_is_deterministic_and_manifest_binds_exact_ assert manifest["schema_version"] == 1 assert manifest["project"] == {"id": str(project_id), "name": project.name} assert manifest["required_variables"] == [] + assert "required_connections" not in manifest assert [entry["id"] for entry in manifest["flows"]] == [str(first_id), str(second_id)] for entry in manifest["flows"]: + assert "required_connections" not in entry payload = archive.read(entry["path"]) assert entry["sha256"] == hashlib.sha256(payload).hexdigest() assert entry["size"] == len(payload) @@ -497,6 +499,98 @@ async def test_build_project_artifact_preserves_variable_references_and_lists_re assert manifest["flows"][0]["required_variables"] == ["MY_INTERNAL_API_URL", "OPENAI_API_KEY"] +@pytest.mark.asyncio +async def test_build_project_artifact_lists_required_connections_and_scopes() -> None: + actor_id = uuid4() + project_id = uuid4() + project = Folder(id=project_id, name="Connection references", user_id=actor_id) + flow = _flow(owner_id=actor_id, project_id=project_id) + flow.data = { + "nodes": [ + { + "data": { + "node": { + "template": { + "calendar": { + "name": "calendar", + "type": "connection_ref", + "provider": "google_workspace", + "value": "google_workspace/work", + "required_scopes": ["calendar.readonly", "userinfo.email"], + }, + "drive": { + "name": "drive", + "type": "connection_ref", + "provider": "google_workspace", + "value": "google_workspace/work", + "required_scopes": ["drive.readonly"], + }, + } + } + } + } + ], + "edges": [], + } + session = _session_with_flows([flow]) + user = SimpleNamespace(id=actor_id, is_superuser=False) + + artifact, *_ = await _build_authorized(session=session, user=user, project=project) + + assert artifact.flows[0].required_connections[0].provider == "google_workspace" + assert artifact.flows[0].required_connections[0].name == "work" + assert artifact.flows[0].required_connections[0].scopes == ( + "calendar.readonly", + "drive.readonly", + "userinfo.email", + ) + with zipfile.ZipFile(io.BytesIO(artifact.content)) as archive: + exported = json.loads(archive.read(f"flows/{flow.id}.json")) + manifest = json.loads(archive.read("manifest.json")) + assert manifest["schema_version"] == 4 + assert manifest["required_connections"] == [ + { + "provider": "google_workspace", + "name": "work", + "scopes": ["calendar.readonly", "drive.readonly", "userinfo.email"], + } + ] + assert manifest["flows"][0]["required_connections"] == manifest["required_connections"] + calendar = exported["data"]["nodes"][0]["data"]["node"]["template"]["calendar"] + assert calendar["value"] == "google_workspace/work" + + +def test_normalized_flow_rejects_connection_provider_mismatch() -> None: + from langflow.services.deployment_artifacts import builder + + snapshot = builder._FlowSnapshot( + flow_id=uuid4(), + name="Provider mismatch", + payload={ + "data": { + "nodes": [ + { + "data": { + "node": { + "template": { + "connection": { + "type": "connection_ref", + "provider": "slack", + "value": "google_workspace/work", + } + } + } + } + } + ] + } + }, + ) + + with pytest.raises(ProjectArtifactError, match="does not match its declared provider"): + builder._normalized_flow_bytes(snapshot) + + @pytest.mark.asyncio async def test_build_project_artifact_keeps_newline_heavy_code_as_bounded_string() -> None: actor_id = uuid4() @@ -611,7 +705,7 @@ def test_secret_scrub_uses_bounded_memory_for_wide_deep_structured_value() -> No tracemalloc.start() try: - content, _ = builder._normalized_flow_bytes(snapshot) + content, _, _ = builder._normalized_flow_bytes(snapshot) _, peak = tracemalloc.get_traced_memory() finally: tracemalloc.stop() @@ -670,9 +764,10 @@ def test_normalized_flow_bytes_accepts_model_valid_sparse_data(data: object, exp ) original_payload = deepcopy(snapshot.payload) - content, required_variables = builder._normalized_flow_bytes(snapshot) + content, required_variables, required_connections = builder._normalized_flow_bytes(snapshot) assert json.loads(content) == {"data": expected_data} assert required_variables == () + assert required_connections == () assert snapshot.payload == original_payload @@ -702,7 +797,7 @@ def test_normalized_flow_bytes_does_not_mutate_shared_flow_data() -> None: payload={"data": flow.data}, ) - content, _ = builder._normalized_flow_bytes(snapshot) + content, _, _ = builder._normalized_flow_bytes(snapshot) exported = json.loads(content) assert exported["data"]["nodes"][0]["data"]["node"]["template"]["password"]["value"] is None diff --git a/src/backend/tests/unit/utils/test_flow_secrets.py b/src/backend/tests/unit/utils/test_flow_secrets.py index c67b2424ac2f..3591d6cbe2ce 100644 --- a/src/backend/tests/unit/utils/test_flow_secrets.py +++ b/src/backend/tests/unit/utils/test_flow_secrets.py @@ -25,6 +25,30 @@ def test_default_scrub_still_nulls_variable_references() -> None: assert _template(flow_data)["api_key"]["value"] == "OPENAI_API_KEY" +def test_scrub_preserves_only_valid_connection_references() -> None: + flow_data = _flow_data( + { + "valid": { + "name": "valid", + "type": "connection_ref", + "password": True, + "value": "google_workspace/work", + }, + "invalid": { + "name": "invalid", + "type": "connection_ref", + "password": True, + "value": "not a handle", + }, + } + ) + + stripped = strip_secret_field_values(flow_data) + + assert _template(stripped)["valid"]["value"] == "google_workspace/work" + assert _template(stripped)["invalid"]["value"] is None + + def test_preserving_scrub_keeps_and_collects_variable_references() -> None: flow_data = _flow_data( { diff --git a/src/lfx/src/lfx/services/authorization/base.py b/src/lfx/src/lfx/services/authorization/base.py index fbcbfc574c1d..dbbd72174108 100644 --- a/src/lfx/src/lfx/services/authorization/base.py +++ b/src/lfx/src/lfx/services/authorization/base.py @@ -30,6 +30,7 @@ class AuthzContext(TypedDict, total=False): file_user_id: _UUID | None share_user_id: _UUID | None provider_account_user_id: _UUID | None + connection_owner_id: _UUID | None voice_user_id: _UUID | None workspace_id: _UUID | None folder_id: _UUID | None