diff --git a/.secrets.baseline b/.secrets.baseline index 476b6ebdd0ab..8d5d62ed1d91 100644 --- a/.secrets.baseline +++ b/.secrets.baseline @@ -6812,7 +6812,7 @@ "filename": "src/lfx/src/lfx/inputs/input_mixin.py", "hashed_secret": "3442496b96dd01591a8cd44b1eec1368ab728aba", "is_verified": false, - "line_number": 21, + "line_number": 23, "is_secret": false } ], diff --git a/BUNDLE_API.md b/BUNDLE_API.md index 15d9c8e9853f..ea0288ee24d5 100644 --- a/BUNDLE_API.md +++ b/BUNDLE_API.md @@ -43,6 +43,21 @@ that does not list `str(BUNDLE_API_VERSION)` is rejected at install time with | `DictInput` / `NestedDictInput` | `lfx.io` | | `FileInput` / `LinkInput` | `lfx.io` | | `HandleInput` | `lfx.io` | +| `ConnectionInput` (legacy Composio connection flow) | `lfx.inputs` | +| `ConnectionRefInput` (portable host-resolved reference) | `lfx.io` | + +### Integrations + +| Symbol | Source | +| --- | --- | +| `ConnectionRef` / `ResolvedCredential` / `CredentialLease` | `lfx.integrations` | +| `ConnectionResolutionRequest` / `ConnectionStatus` | `lfx.integrations` | +| `IntegrationError` and typed subclasses / `INTEGRATION_ERROR_CODES` | `lfx.integrations` | +| `normalize_integration_error()` / `register_error_normalizer()` | `lfx.integrations` | +| `IntegrationProvider` / `OAuthProfile` / `IntegrationCapability` / `ScopeSet` | `lfx.integrations` | +| `integration_action()` | `lfx.integrations` | +| `Component.resolve_connection(field_name)` | `lfx.custom.custom_component.component.Component` | +| `BaseConnectionResolverService`, `ConnectionAccessPolicy` | `lfx.services.connection` | ### Outputs @@ -71,6 +86,7 @@ that does not list `str(BUNDLE_API_VERSION)` is rejected at install time with | Manifest schema (`extension.json` / `[tool.langflow.extension]`) | `lfx.extension.manifest.ExtensionManifest` | | `BundleRef` (one entry in optional `bundles[]`; bundle names must be unique) | `lfx.extension.manifest.BundleRef` | | `ProviderManifestEntry` (one entry in the optional `providers[]`) | `lfx.extension.manifest.ProviderManifestEntry` | +| `IntegrationProvider` entries in optional `integrations[]` | `lfx.integrations.IntegrationProvider` | | `LfxCompat` (declared as `manifest.lfx`) | `lfx.extension.manifest.LfxCompat` | | `BUNDLE_API_VERSION` (the integer this lfx ships) | `lfx.extension.manifest` | | `EXTENSION_SCHEMA_URL` / `SCHEMA_VERSION` | `lfx.extension.manifest` | @@ -189,7 +205,28 @@ the deserialize half is covered by ### v0 (this release) +- Enforced the unreleased connection resolver contract through a final `resolve` + entry point. Hosts now implement `_get_access_policy` and `_resolve`; ownership + and non-interactive/share checks run before credential access. Required scopes + reject unverified credentials with a typed diagnostic. Resolution failures carry + fixed reason codes and actionable guidance without raw credential values or + exception chains. `run_flow` now activates its injected variables and environment + policy for credential lookups, restoring the prior scope after execution. + No previously released API changes; `BUNDLE_API_VERSION` remains 1. - Initial surface enumerated above. Frozen as `BUNDLE_API_VERSION = 1`. +- Added the provider-neutral connection-reference, resolver, capability, + integration-error, and telemetry contracts used by dedicated integration + bundles. This is an additive surface change and does not change + `BUNDLE_API_VERSION`. +- Hardened the new connection contracts before their first release: integration + exports load lazily; `ScopeSet.covers` requires an explicit `provider` and shares + normalization with resolvers through `ScopeSet.missing`; leases activate + conditional scopes and support pre-run construction. Malformed credentials + produce sanitized typed errors; wrapped HTTP failures preserve their status, + and ambiguous 403s no longer claim missing scopes. The authorization floor + accepts an optional host-verified `explicit_share_authorized` decision for + actor ownership mismatches without bypassing other denies. No previously + released Bundle API signature changes; `BUNDLE_API_VERSION` remains 1. - `ExtensionManifest.version` now accepts the canonical PEP 440 stable, dev, alpha, beta, and release-candidate forms emitted by the repository's bundle release pipeline, in addition to the existing SemVer 2.0.0 forms. Runtime diff --git a/design/dedicated-integrations/connection-contract.md b/design/dedicated-integrations/connection-contract.md index 2b008e8a5758..da465296adbb 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-10 (resolver authorization and credential diagnostics 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 @@ -124,6 +124,20 @@ string is simpler for tweaks, env, and manifest sorting; the dict is the parsed is a frozen dataclass `{ref, principal: ExecutionPrincipal, required_scopes: frozenset[str], component_id, flow_id, run_id}`; optional `async def describe(self, ref, principal) -> ConnectionStatus | None` for pickers and health (default `None`). +- `resolve()` is the base-owned, final entry point; subclass creation rejects direct or inherited overrides. + Hosts implement two abstract hooks: `_get_access_policy(request) -> ConnectionAccessPolicy` and + `_resolve(request, policy) -> ResolvedCredential`. The first loads only non-secret ownership/opt-in metadata + and any host-verified share decision. The base rejects unknown/anonymous principals before that lookup and + applies the complete portable deny floor before the credential hook can decrypt or refresh anything. + Overriding `authorize_principal()` cannot weaken this path. The base also checks required scope coverage + before returning a credential. These runtime checks apply to the persistent, OAuth and Enterprise resolvers; + they do not depend on each implementation remembering a helper or a route-matrix entry. + Existing unreleased host resolvers must move their credential logic from `resolve` into `_resolve` and add + the policy hook. `ConnectionAccessPolicy` is immutable, strictly validated host data: `owner_kind`, + `connection_owner_id`, optional `connection_id`, `allow_non_interactive=False` and + `explicit_share_authorized=False`. Hosts must resolve the same connection that was authorized, and guard + against ownership/policy changes between the metadata lookup and secret access. Host Python code remains + trusted; this contract prevents omitted checks, not malicious plugins or forged host metadata. - The member is added to `src/lfx/src/lfx/services/schema.py` and `src/backend/base/langflow/services/schema.py`. `deps.get_connection_resolver()` follows the `get_checkpoint_service()` pattern (`src/lfx/src/lfx/services/deps.py:204`): registered service, else the built-in @@ -142,6 +156,18 @@ string is simpler for tweaks, env, and manifest sorting; the dict is the parsed (`helpers/flow.py:580-611`). A `user` owner kind without an owner id fails closed; host implementations must not treat missing ownership metadata as an implicit match. + `authorize_principal` accepts the optional keyword `explicit_share_authorized=False`. + A host may set it only after authorizing `connection:execute` for an actor on a route + family that permits shares. It satisfies only an owner-id mismatch; unknown/anonymous + principals, missing actor or owner metadata, environment restrictions and non-interactive + opt-in still apply. The decision must never be accepted from flow JSON or component inputs. + + Discovery currently runs lazily on the first non-settings service lookup. A broken + configured resolver aborts discovery and subsequent lookups continue to fail closed; + the environment fallback is never selected. Hosts that require failure before accepting + requests must invoke `ServiceManager.discover_plugins()` during startup. A separate + fallback cache permits later host registration and is disposed with the service manager. + Rejected: piggybacking on `VARIABLE_SERVICE` (string-only; the DB variant has no share semantics; a variable named `LF_CONNECTION__X` could impersonate a connection in DB mode); a callable in `graph.context` (not picklable, copied by `_copy_graph`, invisible to the CI matrices); a method on `BaseAuthorizationService` (the OSS pass-through may be @@ -190,12 +216,16 @@ authorization. Deferred webhook setup must adopt the same preflight when impleme **Decision: one `EnvConnectionResolver` in lfx serves `lfx run`, embedded Python, and `lfx serve`, because the serve request scope is already a ContextVar the variable service reads.** -- `lfx/services/connection/env_resolver.py`: `resolve()` computes `ref.env_key()` and calls +- `lfx/services/connection/env_resolver.py`: `_get_access_policy()` supplies environment ownership; + `_resolve()` computes `ref.env_key()` and calls `get_variable_service().get_variable(key)`. That one call already implements request scope (`activate_request_variables` in `src/lfx/src/lfx/cli/common.py:447-449`, `LANGFLOW_REQUEST_VARIABLES` JSON via `runtime_variables.py`, the `x-langflow-global-var-*` alias), then `safe_getenv` with reserved names denied, skipped under `no_env_fallback`. No new ContextVar. If the owners want the ticket's two names, `RequestScopedConnectionResolver` is a trivial subclass (question 12.a.1). +- `run_flow` activates its graph's request-variable and no-environment-fallback ContextVars for execution, + including human-input runs, then restores both in `finally`, matching serve execution. Supplied request + credentials therefore take precedence over ambient process credentials and cannot leak into later runs. - Trust boundary: in standalone lfx the request scope is the intended injection channel, not a bypass. There is no database, no user, and no connection-provisioning permission to enforce; the only principal is the serve caller, who is authenticated by the serve API key and already controls the flow's inputs. A caller can substitute only a @@ -210,9 +240,21 @@ request scope is already a ContextVar the variable service reads.** `{"access_token", "token_type", "expires_at", "scopes", "account": {"id", "display", "tenant_id"}}`; `normalize_parsed_variables` (`request_scope.py:28`) already serializes nested JSON, so detection is "starts with `{`". Refresh is the injector's job (`refreshable=False`). +- When an action declares any `required_scopes`, bare tokens and JSON without `scopes` fail with + `ScopeMissingError`, `details.scopes_verified=False`, and an instruction to supply scope metadata. + Verified but insufficient scopes fail with the same error code and `details.scopes_verified=True`. + Bare tokens remain supported for actions with no declared scopes. For headless injection, "verified" means + the operator supplied scope metadata; lfx does not introspect the token at the provider. - Failure: `ConnectionUnresolvedError` names the handle, the env key, and the JSON form, never a value. `lfx run` gains `validate_connection_refs_for_env` beside `validate_global_variables_for_env` so the run fails before execution under `--check-variables`. `lfx serve` surfaces the typed error through the normal component-error path. +- Resolution errors expose a fixed `reason` in both the attribute and `details`, plus source-authored guidance: + `missing`, `env-fallback-disabled`, `malformed-json`, `long-lived-secret`, `unsupported-fields`, + `invalid-access-token`, `invalid-scopes`, `invalid-token-type`, `invalid-account`, `invalid-expiry`, or + `invalid-credential`. Long-lived-secret guidance names only the static forbidden field list; unsupported + field names and values are never echoed. Public parsing errors are raised outside the exception handler, + retaining neither the raw exception cause nor context. With `no_env_fallback`, runtime and CLI preflight + direct the operator to request-scoped injection or the host secret provider instead of setting process env. - Identity: the `serve_identity.py` label becomes `ExecutionPrincipal(kind="headless_operator", actor_label=...)`; the `run/_defaults.py` throwaway UUID maps to the same kind. The resolver treats both as instance-or-environment only. @@ -301,9 +343,11 @@ now.** risk: read | write | destructive, component_ref, mcp_tool}`. `ConditionalScopeRequirement{scope, role, condition}` preserves `optional` versus `alternative`; `ScopeCondition{kind: input_present | input_truthy, input}` is evaluated against the action's declared input schema. The matrix checker rejects a condition that - names a missing input. `ScopeSet.covers(capability, inputs, granted) -> missing` first activates conditional - requirements, then performs provider-aware normalization (Google URL scopes, Graph short names, Slack bot versus - user scopes). The picker and resolver therefore apply the same executable rule instead of interpreting prose. + names a missing input. `ScopeSet.covers(capability, inputs, granted, provider=...) -> missing` first activates + conditional requirements, then calls `ScopeSet.missing(provider=..., required=..., granted=...)` for the same + normalization used by the resolver. Google URL scopes (provider ids `google` and `google_workspace`) and Graph + short names are normalized. Slack bot/user identity remains a separate auth-profile check; their scope names + are never treated as interchangeable. The provider is explicit because capability ids need not be qualified. - Capability ids are the matrices' `action_id` values. `required` rows become `required_scopes`; `optional` and `alternative` rows become `conditional_scopes` without losing their role or predicate. The capability's `auth_profile_id` and `identity` must match the selected connection before scope coverage is evaluated. diff --git a/scripts/migrate/check_bundle_api_changelog.py b/scripts/migrate/check_bundle_api_changelog.py index 94972eaff42f..09c6cc0e619a 100755 --- a/scripts/migrate/check_bundle_api_changelog.py +++ b/scripts/migrate/check_bundle_api_changelog.py @@ -64,6 +64,13 @@ # The package facade (re-exports define the surface) "src/lfx/src/lfx/extension/__init__.py", "src/lfx/src/lfx/extension/validate.py", + # Dedicated-integration public surface + "src/lfx/src/lfx/integrations", + "src/lfx/src/lfx/inputs/input_mixin.py", + "src/lfx/src/lfx/inputs/inputs.py", + "src/lfx/src/lfx/io/__init__.py", + "src/lfx/src/lfx/custom/custom_component/component.py", + "src/lfx/src/lfx/services/connection", ) diff --git a/src/backend/base/langflow/services/factory.py b/src/backend/base/langflow/services/factory.py index a9c4e170bc1e..564b66a4c32a 100644 --- a/src/backend/base/langflow/services/factory.py +++ b/src/backend/base/langflow/services/factory.py @@ -78,7 +78,9 @@ 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 in {"mcp_composer", "model_provider_policy", "policy_bundle"}: + if service_name == "connection_resolver": + module_name = "lfx.services.connection.env_resolver" + elif service_name in {"mcp_composer", "model_provider_policy", "policy_bundle"}: module_name = f"lfx.services.{service_name}.service" else: module_name = f"langflow.services.{service_name}.service" diff --git a/src/backend/base/langflow/services/schema.py b/src/backend/base/langflow/services/schema.py index 36a757ba90ac..0f8877ebc53b 100644 --- a/src/backend/base/langflow/services/schema.py +++ b/src/backend/base/langflow/services/schema.py @@ -18,6 +18,7 @@ class ServiceType(str, Enum): TASK_SERVICE = "task_service" STORE_SERVICE = "store_service" VARIABLE_SERVICE = "variable_service" + CONNECTION_RESOLVER_SERVICE = "connection_resolver_service" STORAGE_SERVICE = "storage_service" STATE_SERVICE = "state_service" TRACING_SERVICE = "tracing_service" diff --git a/src/backend/base/langflow/services/telemetry/schema.py b/src/backend/base/langflow/services/telemetry/schema.py index b6937138fef4..073e10548ba4 100644 --- a/src/backend/base/langflow/services/telemetry/schema.py +++ b/src/backend/base/langflow/services/telemetry/schema.py @@ -28,6 +28,18 @@ class DeploymentPayload(BasePayload): wxo_tenant_id: str | None = Field(default=None, serialization_alias="wxoTenantId") +class IntegrationActionPayload(BasePayload): + """Low-cardinality integration action event with no connection identifiers.""" + + provider: str + capability: str + ms: int + success: bool + error_code: str | None = Field(None, serialization_alias="errorCode") + owner_kind: str = Field(serialization_alias="ownerKind") + principal_kind: str = Field(serialization_alias="principalKind") + + class ShutdownPayload(BasePayload): time_running: int = Field(serialization_alias="timeRunning") diff --git a/src/backend/base/langflow/services/telemetry/service.py b/src/backend/base/langflow/services/telemetry/service.py index aa15d37e69ee..bb8a84960d07 100644 --- a/src/backend/base/langflow/services/telemetry/service.py +++ b/src/backend/base/langflow/services/telemetry/service.py @@ -33,6 +33,8 @@ from lfx.services.settings.service import SettingsService from pydantic import BaseModel + from langflow.services.telemetry.schema import IntegrationActionPayload + class TelemetryService(Service): name = "telemetry_service" @@ -119,6 +121,10 @@ async def log_package_run(self, payload: RunPayload) -> None: async def log_package_deployment(self, payload: DeploymentPayload) -> None: await self._queue_event((self.send_telemetry_data, payload, "deployment")) + async def log_integration_action(self, payload: IntegrationActionPayload) -> None: + """Queue an integration event through the normal tracking-consent boundary.""" + await self._queue_event((self.send_telemetry_data, payload, "integration_action")) + async def log_package_deployment_provider(self, payload: DeploymentPayload) -> None: await self._queue_event((self.send_telemetry_data, payload, "deployment_provider")) diff --git a/src/backend/tests/unit/test_telemetry.py b/src/backend/tests/unit/test_telemetry.py index 375e1b795a3d..ec8d0a3f1e69 100644 --- a/src/backend/tests/unit/test_telemetry.py +++ b/src/backend/tests/unit/test_telemetry.py @@ -10,7 +10,7 @@ OpenTelemetry, ThreadSafeSingletonMetaUsingWeakref, ) -from langflow.services.telemetry.schema import DeploymentPayload +from langflow.services.telemetry.schema import DeploymentPayload, IntegrationActionPayload from langflow.services.telemetry.service import TelemetryService @@ -43,6 +43,30 @@ async def test_log_package_deployment(telemetry_service): assert path == "deployment" +@pytest.mark.asyncio +@pytest.mark.parametrize("do_not_track", [False, True]) +async def test_integration_action_uses_telemetry_queue(telemetry_service, do_not_track): + telemetry_service.do_not_track = do_not_track + payload = IntegrationActionPayload( + provider="google", + capability="drive.read", + owner_kind="env", + principal_kind="headless_operator", + ms=1, + success=True, + ) + await telemetry_service.log_integration_action(payload) + if do_not_track: + assert telemetry_service.telemetry_queue.empty() + else: + send, queued, path = telemetry_service.telemetry_queue.get_nowait() + assert send == telemetry_service.send_telemetry_data + assert queued == payload + assert path == "integration_action" + telemetry_service.telemetry_queue.task_done() + await telemetry_service.client.aclose() + + @pytest.mark.asyncio async def test_log_package_deployment_provider(telemetry_service): payload = DeploymentPayload( diff --git a/src/lfx/src/lfx/cli/validation/__init__.py b/src/lfx/src/lfx/cli/validation/__init__.py index e443494d2bd5..4399ce0f1391 100644 --- a/src/lfx/src/lfx/cli/validation/__init__.py +++ b/src/lfx/src/lfx/cli/validation/__init__.py @@ -6,6 +6,7 @@ from lfx.cli.validation._env_validation import ( is_valid_env_var_name, + validate_connection_refs_for_env, validate_global_variables_for_env, ) from lfx.cli.validation.core import ( @@ -56,6 +57,7 @@ "_render_result", "is_valid_env_var_name", "validate_command", + "validate_connection_refs_for_env", "validate_flow_file", "validate_global_variables_for_env", ] diff --git a/src/lfx/src/lfx/cli/validation/_env_validation.py b/src/lfx/src/lfx/cli/validation/_env_validation.py index 976a4abe64dc..7535d375d4b1 100644 --- a/src/lfx/src/lfx/cli/validation/_env_validation.py +++ b/src/lfx/src/lfx/cli/validation/_env_validation.py @@ -1,10 +1,13 @@ """Validation utilities for CLI commands.""" +from __future__ import annotations + import re from typing import TYPE_CHECKING if TYPE_CHECKING: from lfx.graph.graph.base import Graph + from lfx.integrations.errors import ConnectionUnresolvedError def is_valid_env_var_name(name: str) -> bool: @@ -27,7 +30,7 @@ def is_valid_env_var_name(name: str) -> bool: return bool(re.match(pattern, name)) -def validate_global_variables_for_env(graph: "Graph") -> list[str]: +def validate_global_variables_for_env(graph: Graph) -> list[str]: """Validate that all global variables with load_from_db=True can be used as environment variables. When the database is not available (noop mode), global variables with load_from_db=True @@ -70,3 +73,59 @@ def validate_global_variables_for_env(graph: "Graph") -> list[str]: ) return errors + + +def validate_connection_refs_for_env(graph: Graph) -> list[ConnectionUnresolvedError]: + """Return typed failures for connection refs absent from headless injection channels.""" + from lfx.integrations.errors import ConnectionUnresolvedError + 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.variable.request_scope import ( + get_active_request_variables, + is_env_fallback_disabled, + normalize_parsed_variables, + ) + from lfx.utils.env_var_security import safe_getenv + + errors: list[ConnectionUnresolvedError] = [] + request_variables = normalize_parsed_variables( + graph.context.get("request_variables") or get_active_request_variables() or {} + ) + no_env_fallback = bool(graph.context.get("no_env_fallback")) or is_env_fallback_disabled() + uses_environment: bool | None = None + + for vertex in graph.vertices: + template = vertex.data.get("node", {}).get("template", {}) + for field_name, field in template.items(): + if not isinstance(field, dict) or field.get("type") != "connection_ref": + continue + value = vertex.params.get(field_name, field.get("value")) + if not value: + continue + try: + ref = ConnectionRef.parse(value) + except ValueError: + errors.append(ConnectionUnresolvedError("")) + continue + if uses_environment is None: + # A configured subclass may use other credential stores, so only + # the built-in implementation has this environment requirement. + uses_environment = type(get_connection_resolver()) is EnvConnectionResolver + if not uses_environment: + continue + env_key = ref.env_key() + alias = f"x-langflow-global-var-{env_key.lower().replace('_', '-')}" + injected = request_variables.get(env_key) or request_variables.get(alias) + if not injected and not no_env_fallback: + injected = safe_getenv(env_key) or safe_getenv(alias) + if not injected: + errors.append( + ConnectionUnresolvedError( + ref.to_handle(), + env_key=env_key, + provider=ref.provider, + reason="env-fallback-disabled" if no_env_fallback else "missing", + ) + ) + return errors diff --git a/src/lfx/src/lfx/custom/custom_component/component.py b/src/lfx/src/lfx/custom/custom_component/component.py index 6a77c15c3081..b3cde1da8f3b 100644 --- a/src/lfx/src/lfx/custom/custom_component/component.py +++ b/src/lfx/src/lfx/custom/custom_component/component.py @@ -57,6 +57,7 @@ from lfx.graph.edge.schema import EdgeData from lfx.graph.vertex.base import Vertex from lfx.inputs.inputs import InputTypes + from lfx.integrations.models import CredentialLease from lfx.schema.dataframe import DataFrame from lfx.schema.log import LoggableType from lfx.services.model_provider_policy import ModelProviderPolicyPurpose @@ -604,6 +605,45 @@ def get_input(self, name: str) -> Any: msg = f"Input {name} not found in {self.__class__.__name__}" raise ValueError(msg) + def resolve_connection(self, field_name: str) -> CredentialLease: + """Create a lazy credential lease for a declared connection-reference input.""" + from lfx.inputs.inputs import ConnectionRefInput + from lfx.integrations.models import ConnectionRef, ConnectionResolutionRequest, CredentialLease + from lfx.services.authorization.base import ExecutionPrincipal + from lfx.services.deps import get_connection_resolver + + input_model = self._inputs.get(field_name) + if not isinstance(input_model, ConnectionRefInput): + msg = f"Input {field_name!r} is not a ConnectionRefInput" + raise TypeError(msg) + value = getattr(self, field_name, input_model.value) + ref = ConnectionRef.parse(value) + if ref.provider != input_model.provider: + msg = ( + f"Connection reference provider {ref.provider!r} does not match " + f"declared provider {input_model.provider!r}" + ) + raise ValueError(msg) + graph = getattr(self, "graph", None) + principal = getattr(graph, "execution_principal", ExecutionPrincipal.unknown()) + required_scopes = set(input_model.required_scopes) + inputs = {name: getattr(self, name, model.value) for name, model in self._inputs.items()} + required_scopes.update( + requirement.scope + for requirement in input_model.conditional_scopes + if requirement.condition.is_active(inputs) + ) + run_id = getattr(graph, "_run_id", None) + request = ConnectionResolutionRequest( + ref=ref, + principal=principal, + required_scopes=frozenset(required_scopes), + component_id=self.get_id(), + flow_id=str(graph.flow_id) if graph is not None and graph.flow_id is not None else None, + run_id=str(run_id) if run_id else None, + ) + return CredentialLease(get_connection_resolver(), request) + def get_output(self, name: str) -> Any: """Retrieves the output with the specified name. diff --git a/src/lfx/src/lfx/extension/manifest.py b/src/lfx/src/lfx/extension/manifest.py index 5c206cc9f089..27397511583b 100644 --- a/src/lfx/src/lfx/extension/manifest.py +++ b/src/lfx/src/lfx/extension/manifest.py @@ -49,6 +49,9 @@ model_validator, ) +from lfx.integrations.capabilities import IntegrationProvider +from lfx.integrations.models import provider_env_segment + # --------------------------------------------------------------------------- # Constants # --------------------------------------------------------------------------- @@ -480,7 +483,7 @@ class ExtensionManifest(BaseModel): Required fields: - id, version, name, bundles, lfx Optional: - - description, capabilities, schema (``$schema``) + - description, capabilities, integrations, schema (``$schema``) Deferred (rejected with ``field-deferred-in-this-milestone`` when set): - services, routes, hooks, starter_projects, userConfig """ @@ -553,6 +556,11 @@ class ExtensionManifest(BaseModel): description="Optional declared capabilities (v0: requiresCredentials only).", ) + integrations: tuple[IntegrationProvider, ...] = Field( + default=(), + description="Provider authentication profiles and executable integration capabilities.", + ) + # ------------------------------------------------------------------ # Deferred fields. We model them as ``None``-only so that downstream # tooling can distinguish "absent" from "explicitly set to a value the @@ -614,6 +622,18 @@ def _validate_provider_name_uniqueness(self) -> ExtensionManifest: raise ValueError(msg) return self + @model_validator(mode="after") + def _validate_integration_provider_uniqueness(self) -> ExtensionManifest: + provider_ids = [provider.provider_id for provider in self.integrations] + if len(set(provider_ids)) != len(provider_ids): + msg = "Integration provider ids must be unique within an extension" + raise ValueError(msg) + env_segments = [provider_env_segment(provider_id) for provider_id in provider_ids] + if len(set(env_segments)) != len(env_segments): + msg = "Integration provider ids must map to unique environment-key segments" + raise ValueError(msg) + return self + @model_validator(mode="after") def _validate_bundle_uniqueness(self) -> ExtensionManifest: # Bundle names are public registry and saved-flow namespaces, so they diff --git a/src/lfx/src/lfx/graph/graph/base.py b/src/lfx/src/lfx/graph/graph/base.py index 0da73cd5e8a2..fb20b8cd170d 100644 --- a/src/lfx/src/lfx/graph/graph/base.py +++ b/src/lfx/src/lfx/graph/graph/base.py @@ -48,6 +48,7 @@ ) from lfx.schema.dotdict import dotdict from lfx.schema.schema import INPUT_FIELD_NAME, InputType, OutputValue +from lfx.services.authorization.base import ExecutionPrincipal from lfx.services.cache.utils import CacheMiss from lfx.services.deps import get_chat_service, get_tracing_service from lfx.utils.async_helpers import run_until_complete @@ -162,6 +163,7 @@ def __init__( self.flow_name = flow_name self.description = description self.user_id = user_id + self.execution_principal = ExecutionPrincipal.unknown() # Warm-registry templates need the parsed graph structure without # executing component constructors at preload/reconcile time. Normal # graphs keep the historical eager-instantiation behavior. @@ -1680,6 +1682,7 @@ def _copy_graph( before_initialize(new_graph) new_graph.requires_extension_event_replay = self.requires_extension_event_replay + new_graph.execution_principal = self.execution_principal # Store the newly created object in memo memo[id(self)] = new_graph @@ -1724,6 +1727,7 @@ def __setstate__(self, state): # Graphs cached before source-flow provenance was introduced remain # loadable and simply have no additional trusted storage namespace. state.setdefault("source_flow_id", None) + state.setdefault("execution_principal", ExecutionPrincipal.unknown()) run_manager = state["run_manager"] if isinstance(run_manager, RunnableVerticesManager): state["run_manager"] = run_manager @@ -3162,6 +3166,7 @@ async def create_subgraph(self, vertex_ids: set[str]) -> AsyncIterator[Graph]: subgraph._tracing_service_initialized = True subgraph._run_id = self._run_id subgraph.session_id = self.session_id + subgraph.execution_principal = self.execution_principal # A subgraph extends the parent's run, so it inherits the ephemeral # (no-persist) decision too. subgraph.persist_messages = self.persist_messages diff --git a/src/lfx/src/lfx/inputs/__init__.py b/src/lfx/src/lfx/inputs/__init__.py index 21f7e00e9faa..fae2286e2be5 100644 --- a/src/lfx/src/lfx/inputs/__init__.py +++ b/src/lfx/src/lfx/inputs/__init__.py @@ -3,6 +3,7 @@ BoolInput, CodeInput, ConnectionInput, + ConnectionRefInput, DataDisplayInput, DataFrameInput, DataInput, @@ -41,6 +42,7 @@ "BoolInput", "CodeInput", "ConnectionInput", + "ConnectionRefInput", "DBProviderInput", "DataDisplayInput", "DataFrameInput", diff --git a/src/lfx/src/lfx/inputs/input_mixin.py b/src/lfx/src/lfx/inputs/input_mixin.py index 80a078b86bf5..da0eebb7576e 100644 --- a/src/lfx/src/lfx/inputs/input_mixin.py +++ b/src/lfx/src/lfx/inputs/input_mixin.py @@ -1,5 +1,5 @@ from enum import Enum -from typing import Annotated, Any +from typing import Annotated, Any, Literal from pydantic import ( BaseModel, @@ -12,6 +12,8 @@ from lfx.field_typing.range_spec import RangeSpec from lfx.inputs.validators import CoalesceBool +from lfx.integrations.capabilities import ConditionalScopeRequirement +from lfx.integrations.models import PROVIDER_ID_PATTERN from lfx.schema.cross_module import CrossModuleModel @@ -27,6 +29,7 @@ class FieldTypes(str, Enum): ACTION_PICKER = "actionPicker" DURATION = "duration" CONNECTION = "connect" + CONNECTION_REF = "connection_ref" AUTH = "auth" FILE = "file" PROMPT = "prompt" @@ -55,6 +58,7 @@ class FieldTypes(str, Enum): FieldTypes.AUTH, FieldTypes.FILE, FieldTypes.CONNECTION, + FieldTypes.CONNECTION_REF, FieldTypes.MCP, } @@ -356,6 +360,17 @@ class ConnectionMixin(BaseModel): """List of dictionaries with metadata for each option.""" +class ConnectionRefMixin(BaseModel): + """Provider and capability metadata for a portable connection reference.""" + + provider: str = Field(pattern=PROVIDER_ID_PATTERN, max_length=120) + auth_profile_id: str = "" + required_scopes: list[str] = Field(default_factory=list) + conditional_scopes: list[ConditionalScopeRequirement] = Field(default_factory=list) + identity_kind: Literal["user", "instance", "any"] = "any" + capabilities: list[str] = Field(default_factory=list) + + class TabMixin(BaseModel): """Mixin for tab input fields that allows a maximum of 3 values, each with a maximum of 20 characters.""" diff --git a/src/lfx/src/lfx/inputs/inputs.py b/src/lfx/src/lfx/inputs/inputs.py index 4bcbb1040996..ad87cc3c7b22 100644 --- a/src/lfx/src/lfx/inputs/inputs.py +++ b/src/lfx/src/lfx/inputs/inputs.py @@ -14,6 +14,7 @@ AuthMixin, BaseInputMixin, ConnectionMixin, + ConnectionRefMixin, DatabaseLoadMixin, DropDownMixin, FieldTypes, @@ -852,6 +853,34 @@ class ConnectionInput(BaseInputMixin, ConnectionMixin, MetadataTraceMixin, ToolM track_in_telemetry: CoalesceBool = False # Never track connection strings (may contain credentials) +class ConnectionRefInput(BaseInputMixin, ConnectionRefMixin, MetadataTraceMixin): + """Portable, non-secret reference resolved by the execution host.""" + + field_type: SerializableFieldTypes = FieldTypes.CONNECTION_REF + password: CoalesceBool = False + load_from_db: CoalesceBool = False + track_in_telemetry: CoalesceBool = False + + @model_validator(mode="before") + @classmethod + def _validate_connection_ref(cls, data: Any) -> Any: + if not isinstance(data, dict): + return data + if data.get("tool_mode"): + msg = "Connection references cannot be exposed as tool-call inputs" + raise ValueError(msg) + value = data.get("value") + if value is not None and value != "": + from lfx.integrations.models import ConnectionRef + + ref = ConnectionRef.parse(value) + provider = data.get("provider") + if provider and ref.provider != provider: + msg = f"Connection reference provider {ref.provider!r} does not match declared provider {provider!r}" + raise ValueError(msg) + return data + + class AuthInput(BaseInputMixin, AuthMixin, MetadataTraceMixin): """Represents an authentication input field. @@ -1074,6 +1103,7 @@ class DefaultPromptField(Input): | MultiselectInput | SortableListInput | ConnectionInput + | ConnectionRefInput | FileInput | FloatInput | HandleInput diff --git a/src/lfx/src/lfx/integrations/__init__.py b/src/lfx/src/lfx/integrations/__init__.py new file mode 100644 index 000000000000..dc3b27599b5e --- /dev/null +++ b/src/lfx/src/lfx/integrations/__init__.py @@ -0,0 +1,116 @@ +"""Public provider-neutral contracts for dedicated integrations.""" + +from importlib import import_module +from typing import TYPE_CHECKING, Any + +if TYPE_CHECKING: + from lfx.integrations.capabilities import ( + ConditionalScopeRequirement, + IntegrationCapability, + IntegrationProvider, + OAuthProfile, + ScopeCondition, + ScopeSet, + ) + from lfx.integrations.errors import ( + INTEGRATION_ERROR_CODES, + ActionUnsupportedError, + AuthExpiredError, + ConnectionNotAuthorizedError, + ConnectionUnresolvedError, + IntegrationError, + ProviderUnavailableError, + RateLimitedError, + ScopeMissingError, + normalize_integration_error, + register_error_normalizer, + ) + from lfx.integrations.models import ( + ConnectionAccount, + ConnectionRef, + ConnectionResolutionRequest, + ConnectionStatus, + CredentialLease, + ResolvedCredential, + ) + from lfx.integrations.telemetry import integration_action + +__all__ = [ + "INTEGRATION_ERROR_CODES", + "ActionUnsupportedError", + "AuthExpiredError", + "ConditionalScopeRequirement", + "ConnectionAccount", + "ConnectionNotAuthorizedError", + "ConnectionRef", + "ConnectionResolutionRequest", + "ConnectionStatus", + "ConnectionUnresolvedError", + "CredentialLease", + "IntegrationCapability", + "IntegrationError", + "IntegrationProvider", + "OAuthProfile", + "ProviderUnavailableError", + "RateLimitedError", + "ResolvedCredential", + "ScopeCondition", + "ScopeMissingError", + "ScopeSet", + "integration_action", + "normalize_integration_error", + "register_error_normalizer", +] + +_MODULES = { + **dict.fromkeys( + ( + "ConditionalScopeRequirement", + "IntegrationCapability", + "IntegrationProvider", + "OAuthProfile", + "ScopeCondition", + "ScopeSet", + ), + "capabilities", + ), + **dict.fromkeys( + ( + "ConnectionAccount", + "ConnectionRef", + "ConnectionResolutionRequest", + "ConnectionStatus", + "CredentialLease", + "ResolvedCredential", + ), + "models", + ), + **dict.fromkeys( + ( + "INTEGRATION_ERROR_CODES", + "ActionUnsupportedError", + "AuthExpiredError", + "ConnectionNotAuthorizedError", + "ConnectionUnresolvedError", + "IntegrationError", + "ProviderUnavailableError", + "RateLimitedError", + "ScopeMissingError", + "normalize_integration_error", + "register_error_normalizer", + ), + "errors", + ), + "integration_action": "telemetry", +} + + +def __getattr__(name: str) -> Any: + """Load runtime contracts only when requested, keeping schema imports light.""" + module = _MODULES.get(name) + if module is None: + msg = f"module {__name__!r} has no attribute {name!r}" + raise AttributeError(msg) + value = getattr(import_module(f"{__name__}.{module}"), name) + globals()[name] = value + return value diff --git a/src/lfx/src/lfx/integrations/capabilities.py b/src/lfx/src/lfx/integrations/capabilities.py new file mode 100644 index 000000000000..b52703c7c339 --- /dev/null +++ b/src/lfx/src/lfx/integrations/capabilities.py @@ -0,0 +1,175 @@ +"""Typed integration capability metadata shared by manifests and resolvers.""" + +from __future__ import annotations + +from typing import Any, Literal + +from pydantic import BaseModel, ConfigDict, Field, StrictStr, model_validator + +from lfx.integrations.models import PROVIDER_ID_PATTERN + +OAuthKind = Literal[ + "oauth2_authorization_code", + "oauth2_client_credentials", + "oauth2_device_code", + "service_account", + "service_account_domain_wide_delegation", + "bot_token_install", + "api_key", +] +IntegrationIdentity = Literal["user_delegated", "bot", "service"] +DeploymentContext = Literal["hosted", "self_managed", "desktop", "headless"] + + +class ScopeCondition(BaseModel): + """Input predicate controlling whether a conditional scope is active.""" + + model_config = ConfigDict(extra="forbid", frozen=True) + + kind: Literal["input_present", "input_truthy"] + input: StrictStr = Field(pattern=r"^[a-z][a-z0-9_]*$") + + def is_active(self, inputs: dict[str, Any]) -> bool: + """Evaluate presence separately from truthiness for conditional scopes.""" + if self.kind == "input_present": + return self.input in inputs and inputs[self.input] is not None + return bool(inputs.get(self.input)) + + +class ConditionalScopeRequirement(BaseModel): + """Scope activated only when its declared input predicate matches.""" + + model_config = ConfigDict(extra="forbid", frozen=True) + + scope: StrictStr = Field(min_length=1) + role: Literal["optional", "alternative"] + condition: ScopeCondition + + +class OAuthProfile(BaseModel): + """One named provider authentication profile.""" + + model_config = ConfigDict(extra="forbid", frozen=True) + + id: StrictStr = Field(pattern=r"^[a-z0-9][a-z0-9_-]*$") + kind: OAuthKind + identity: IntegrationIdentity + authorization_url: StrictStr | None = None + token_url: StrictStr | None = None + supports_pkce: bool = False + supports_refresh: bool = False + scope_separator: StrictStr = " " + default_scopes: tuple[StrictStr, ...] = () + client_type_by_context: dict[DeploymentContext, Literal["confidential", "public", "external"]] = Field( + default_factory=dict + ) + owner_by_context: dict[DeploymentContext, Literal["langflow", "customer", "either"]] = Field(default_factory=dict) + tenant_param: StrictStr | None = None + + +class IntegrationCapability(BaseModel): + """One executable provider action and its credential requirements.""" + + model_config = ConfigDict(extra="forbid", frozen=True) + + id: StrictStr = Field(pattern=r"^[a-z0-9][a-z0-9._-]*$") + display_name: StrictStr = Field(min_length=1) + auth_profile_id: StrictStr = Field(pattern=r"^[a-z0-9][a-z0-9_-]*$") + identity: IntegrationIdentity + required_scopes: tuple[StrictStr, ...] = () + conditional_scopes: tuple[ConditionalScopeRequirement, ...] = () + risk: Literal["read", "write", "destructive"] + component_ref: StrictStr | None = None + mcp_tool: StrictStr | None = None + + @model_validator(mode="after") + def _has_an_execution_target(self) -> IntegrationCapability: + if self.component_ref is None and self.mcp_tool is None: + msg = "An integration capability must declare component_ref or mcp_tool" + raise ValueError(msg) + return self + + +class IntegrationProvider(BaseModel): + """Provider metadata and the capability/profile catalog it exposes.""" + + model_config = ConfigDict(extra="forbid", frozen=True) + + provider_id: StrictStr = Field(pattern=PROVIDER_ID_PATTERN) + display_name: StrictStr = Field(min_length=1) + icon: StrictStr | None = None + auth_profiles: tuple[OAuthProfile, ...] + capabilities: tuple[IntegrationCapability, ...] + docs_url: StrictStr | None = None + + @model_validator(mode="after") + def _references_known_profiles(self) -> IntegrationProvider: + profile_ids = [profile.id for profile in self.auth_profiles] + if len(set(profile_ids)) != len(profile_ids): + msg = f"Integration provider {self.provider_id!r} has duplicate auth profile ids" + raise ValueError(msg) + capability_ids = [capability.id for capability in self.capabilities] + if len(set(capability_ids)) != len(capability_ids): + msg = f"Integration provider {self.provider_id!r} has duplicate capability ids" + raise ValueError(msg) + unknown = sorted({cap.auth_profile_id for cap in self.capabilities} - set(profile_ids)) + if unknown: + msg = f"Integration provider {self.provider_id!r} references unknown auth profiles: {', '.join(unknown)}" + raise ValueError(msg) + identities = {profile.id: profile.identity for profile in self.auth_profiles} + mismatched = sorted( + capability.id + for capability in self.capabilities + if identities[capability.auth_profile_id] != capability.identity + ) + if mismatched: + msg = ( + f"Integration provider {self.provider_id!r} has capabilities whose identity does not match " + f"their auth profile: {', '.join(mismatched)}" + ) + raise ValueError(msg) + return self + + +class ScopeSet: + """Provider-aware scope coverage shared by pickers and resolvers.""" + + @staticmethod + def _normalize(provider: str, scope: str) -> str: + """Compare provider scope aliases without conflating token identities.""" + normalized = scope.strip() + if provider in {"google", "google_workspace"}: + normalized = normalized.removeprefix("https://www.googleapis.com/auth/") + elif provider == "microsoft": + normalized = normalized.removeprefix("https://graph.microsoft.com/") + return normalized.casefold() + + @classmethod + def missing( + cls, *, provider: str, required: set[str] | frozenset[str], granted: set[str] | frozenset[str] + ) -> frozenset[str]: + """Return uncovered scopes, preserving their original spelling in errors. + + Token identity and auth-profile compatibility are separate host checks; + Slack user and bot scope names must never be treated as interchangeable. + """ + normalized_granted = {cls._normalize(provider, scope) for scope in granted} + return frozenset(scope for scope in required if cls._normalize(provider, scope) not in normalized_granted) + + @classmethod + def covers( + cls, + capability: IntegrationCapability, + inputs: dict[str, Any], + granted: set[str] | frozenset[str], + *, + provider: str, + ) -> frozenset[str]: + """Return required active scopes not covered by ``granted``.""" + required = set(capability.required_scopes) + required.update( + requirement.scope + for requirement in capability.conditional_scopes + if requirement.condition.is_active(inputs) + ) + return cls.missing(provider=provider, required=required, granted=granted) diff --git a/src/lfx/src/lfx/integrations/errors.py b/src/lfx/src/lfx/integrations/errors.py new file mode 100644 index 000000000000..de43085561c7 --- /dev/null +++ b/src/lfx/src/lfx/integrations/errors.py @@ -0,0 +1,303 @@ +"""Sanitized, machine-readable failures for provider integrations.""" + +from __future__ import annotations + +import re +from collections.abc import Callable, Iterator +from typing import Any, Literal + +_EMAIL_RE = re.compile(r"(? str: + from lfx.utils.url_redaction import redact_urls_in_text + + return _EMAIL_RE.sub("[redacted-email]", redact_urls_in_text(text)) + + +def _sanitize_details(value: Any) -> Any: + if isinstance(value, str): + return _sanitize(value) + if isinstance(value, dict): + return {_sanitize(str(key)): _sanitize_details(item) for key, item in value.items()} + if isinstance(value, list | tuple | set | frozenset): + return [_sanitize_details(item) for item in value] + return value + + +class IntegrationError(Exception): + """Base error whose string form is always safe for clients and telemetry.""" + + code = "provider-unavailable" + + def __init__( + self, + message: str, + *, + hint: str | None = None, + provider: str | None = None, + retryable: bool = False, + http_status: int | None = None, + safe_message: str | None = None, + details: dict[str, Any] | None = None, + ) -> None: + self.message = _sanitize(message) + self.safe_message = _sanitize(safe_message or message) + self.hint = _sanitize(hint) if hint else None + self.provider = provider + self.retryable = retryable + self.http_status = http_status + self.details = _sanitize_details(details or {}) + super().__init__(self.safe_message) + + +ConnectionUnresolvedReason = Literal[ + "missing", + "env-fallback-disabled", + "malformed-json", + "long-lived-secret", + "unsupported-fields", + "invalid-access-token", + "invalid-scopes", + "invalid-token-type", + "invalid-account", + "invalid-expiry", + "invalid-credential", +] + +_CONNECTION_UNRESOLVED_HINTS: dict[ConnectionUnresolvedReason, str] = { + "missing": "Configure the connection for this execution environment.", + "env-fallback-disabled": ( + "Supply the connection through request-scoped variables or the host's secret provider; " + "environment fallback is disabled." + ), + "malformed-json": "Supply a valid credential JSON object containing access_token.", + "long-lived-secret": ( + "Remove refresh_token, client_secret, and password fields; supply only a short-lived access token " + "and supported credential metadata." + ), + "unsupported-fields": "Use only access_token, token_type, expires_at, scopes, and account in credential JSON.", + "invalid-access-token": "Supply a non-empty string in access_token.", + "invalid-scopes": "Supply scopes as a list of non-empty strings.", + "invalid-token-type": "Supply token_type as a non-empty string.", + "invalid-account": "Supply account as an object with id and optional display and tenant_id strings.", + "invalid-expiry": "Supply expires_at as a valid ISO-8601 string or Unix timestamp.", + "invalid-credential": "Supply a token or a credential JSON object with valid metadata.", +} + + +class ConnectionUnresolvedError(IntegrationError): + code = "connection-unresolved" + + def __init__( + self, + handle: str, + *, + env_key: str | None = None, + provider: str | None = None, + reason: ConnectionUnresolvedReason = "missing", + ) -> None: + if reason not in _CONNECTION_UNRESOLVED_HINTS: + msg = "Unknown connection resolution reason" + raise ValueError(msg) + hint = _CONNECTION_UNRESOLVED_HINTS[reason] + if reason == "missing" and env_key: + hint = f"Set {env_key} to a token or credential JSON object." + super().__init__( + f"Connection {handle!r} could not be resolved. {hint}", + hint=hint, + provider=provider, + details={"reason": reason}, + ) + self.handle = handle + self.env_key = env_key + self.reason = reason + + +class ConnectionNotAuthorizedError(IntegrationError): + code = "connection-not-authorized" + + def __init__(self, *, provider: str | None = None, reason: Literal["principal", "provider"] = "principal") -> None: + super().__init__( + "The provider denied this action." + if reason == "provider" + else "This execution principal is not authorized to use the requested connection.", + hint="Check the provider's access and administrator policy." + if reason == "provider" + else "Use an owned or explicitly shared connection.", + provider=provider, + http_status=403, + ) + + +class AuthExpiredError(IntegrationError): + code = "auth-expired" + + def __init__(self, *, provider: str | None = None, http_status: int | None = 401) -> None: + super().__init__( + "The provider credential is expired or was rejected.", + hint="Reconnect the integration and try again.", + provider=provider, + http_status=http_status, + ) + + +class ScopeMissingError(IntegrationError): + code = "scope-missing" + + def __init__( + self, + missing: frozenset[str] = frozenset(), + *, + provider: str | None = None, + scopes_verified: bool = True, + ) -> None: + super().__init__( + "The connection does not grant every scope required by this action." + if scopes_verified + else "The connection's granted scopes are unverified; this action requires verified scope metadata.", + hint="Grant the missing scopes and reconnect." + if scopes_verified + else "Supply credential JSON with scopes, or use a host resolver that verifies granted scopes.", + provider=provider, + http_status=403, + details={"missing": sorted(missing), "scopes_verified": scopes_verified}, + ) + self.missing = missing + + +class RateLimitedError(IntegrationError): + code = "rate-limited" + + def __init__( + self, + *, + provider: str | None = None, + retry_after: float | None = None, + http_status: int | None = 429, + ) -> None: + super().__init__( + "The provider rate limit was reached.", + hint="Retry after the provider's backoff interval.", + provider=provider, + retryable=True, + http_status=http_status, + details={"retry_after": retry_after} if retry_after is not None else None, + ) + self.retry_after = retry_after + + +class ProviderUnavailableError(IntegrationError): + code = "provider-unavailable" + + def __init__(self, *, provider: str | None = None, http_status: int | None = None) -> None: + super().__init__( + "The provider is temporarily unavailable.", + hint="Retry the action later.", + provider=provider, + retryable=True, + http_status=http_status, + ) + + +class ActionUnsupportedError(IntegrationError): + code = "action-unsupported" + + def __init__(self, *, provider: str | None = None, http_status: int | None = None) -> None: + super().__init__( + "The provider does not support this action.", + provider=provider, + http_status=http_status, + ) + + +ErrorNormalizer = Callable[[BaseException], IntegrationError | None] +_NORMALIZERS: dict[str, ErrorNormalizer] = {} + + +def register_error_normalizer(provider: str, normalizer: ErrorNormalizer) -> None: + """Register a bundle-owned SDK error normalizer without importing its SDK in lfx.""" + if not provider or not callable(normalizer): + msg = "provider must be non-empty and normalizer must be callable" + raise ValueError(msg) + _NORMALIZERS[provider] = normalizer + + +def _retry_after(exc: BaseException) -> float | None: + response = getattr(exc, "response", None) + headers = getattr(response, "headers", None) + if headers is None: + return None + raw = headers.get("retry-after") + try: + return float(raw) if raw is not None else None + except (TypeError, ValueError): + return None + + +def _iter_errors(error: BaseException) -> Iterator[BaseException]: + """Visit wrappers, group members, causes and contexts once, including cycles.""" + pending = [error] + seen: set[int] = set() + while pending: + current = pending.pop() + if id(current) in seen: + continue + seen.add(id(current)) + yield current + if current.__context__ is not None: + pending.append(current.__context__) + if current.__cause__ is not None: + pending.append(current.__cause__) + # Supports both built-in ExceptionGroup and its Python 3.10 backport. + pending.extend(reversed(getattr(current, "exceptions", ()))) + + +def normalize_integration_error(exc: BaseException, *, provider: str) -> IntegrationError: + """Map provider/transport failures into the stable sanitized error vocabulary.""" + from lfx.base.mcp.util import extract_http_status + + normalizer = _NORMALIZERS.get(provider) + for error in _iter_errors(exc): + if isinstance(error, IntegrationError): + return error + if normalizer is not None: + normalized = normalizer(error) + if normalized is not None: + return normalized + if getattr(error, "exceptions", None): + continue # Inspect each leaf so status and Retry-After come from the same response. + status = extract_http_status(error) + if status == HTTP_UNAUTHORIZED: + return AuthExpiredError(provider=provider, http_status=status) + if status == HTTP_FORBIDDEN: + headers = getattr(getattr(error, "response", None), "headers", {}) + challenge = headers.get("www-authenticate", "") + if re.search(r'\berror\s*=\s*"?insufficient_scope\b', challenge, re.IGNORECASE): + return ScopeMissingError(provider=provider) + return ConnectionNotAuthorizedError(provider=provider, reason="provider") + if status == HTTP_TOO_MANY_REQUESTS: + return RateLimitedError(provider=provider, retry_after=_retry_after(error), http_status=status) + if status in {HTTP_NOT_FOUND, HTTP_METHOD_NOT_ALLOWED, HTTP_NOT_IMPLEMENTED}: + return ActionUnsupportedError(provider=provider, http_status=status) + if status is not None: + return ProviderUnavailableError(provider=provider, http_status=status) + return ProviderUnavailableError(provider=provider) diff --git a/src/lfx/src/lfx/integrations/models.py b/src/lfx/src/lfx/integrations/models.py new file mode 100644 index 000000000000..9de8bef33c91 --- /dev/null +++ b/src/lfx/src/lfx/integrations/models.py @@ -0,0 +1,204 @@ +"""Provider-neutral connection references and short-lived credential leases.""" + +from __future__ import annotations + +import asyncio +import re +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone +from typing import TYPE_CHECKING, Literal + +from pydantic import BaseModel, ConfigDict, Field, SecretStr, StrictStr + +if TYPE_CHECKING: + from collections.abc import Callable + + from lfx.integrations.errors import AuthExpiredError + from lfx.services.authorization.base import ExecutionPrincipal + from lfx.services.interfaces import ConnectionResolverProtocol + + +PROVIDER_ID_PATTERN = r"^[a-z0-9][a-z0-9._-]*$" +CONNECTION_NAME_PATTERN = r"^[a-z0-9]+(?:_[a-z0-9]+)*$" +_PROVIDER_ID_RE = re.compile(PROVIDER_ID_PATTERN) +_CONNECTION_NAME_RE = re.compile(CONNECTION_NAME_PATTERN) +_ENV_SEPARATOR = "__" +_ENV_PREFIX = "LF_CONNECTION__" +_EXPIRY_MARGIN = timedelta(seconds=60) + + +def provider_env_segment(provider_id: str) -> str: + """Return a collision-free environment-key segment for a provider id. + + Alphanumeric characters are uppercased and punctuation is escaped with its + ASCII hex value. This keeps the key shell-friendly while preserving the + distinction between provider ids such as ``a.b``, ``a-b``, and ``a_b``. + """ + if not _PROVIDER_ID_RE.fullmatch(provider_id): + msg = f"Invalid integration provider id: {provider_id!r}" + raise ValueError(msg) + return "".join(character.upper() if character.isalnum() else f"_{ord(character):02X}" for character in provider_id) + + +class ConnectionRef(BaseModel): + """Portable, non-secret reference stored in flow JSON as ``provider/name``.""" + + model_config = ConfigDict(extra="forbid", frozen=True) + + provider: StrictStr = Field(pattern=PROVIDER_ID_PATTERN, max_length=120) + name: StrictStr = Field(pattern=CONNECTION_NAME_PATTERN, max_length=64) + + @classmethod + def parse(cls, value: str | ConnectionRef) -> ConnectionRef: + """Parse a connection handle, rejecting ambiguous or malformed values.""" + if isinstance(value, cls): + return value + if not isinstance(value, str) or value.count("/") != 1: + msg = "Connection references must use the form '/'" + raise ValueError(msg) + provider, name = value.split("/", 1) + return cls(provider=provider, name=name) + + def to_handle(self) -> str: + """Serialize this reference to its stable flow representation.""" + return f"{self.provider}/{self.name}" + + def env_key(self) -> str: + """Return the environment/request-scope key used by headless runtimes.""" + return f"{_ENV_PREFIX}{provider_env_segment(self.provider)}{_ENV_SEPARATOR}{self.name.upper()}" + + def __str__(self) -> str: + return self.to_handle() + + +class ConnectionAccount(BaseModel): + """Non-secret provider account metadata associated with a credential.""" + + model_config = ConfigDict(extra="forbid", frozen=True) + + id: StrictStr + display: StrictStr | None = None + tenant_id: StrictStr | None = None + + +@dataclass(frozen=True, slots=True, repr=False) +class ResolvedCredential: + """Short-lived credential returned by a host resolver. + + The object refuses pickling so access tokens cannot enter graph snapshots, + background-job payloads, or process caches by accident. + """ + + access_token: SecretStr + token_type: str = "Bearer" # noqa: S105 - OAuth token scheme, not a credential + expires_at: datetime | None = None + granted_scopes: frozenset[str] = frozenset() + scopes_verified: bool = False + account: ConnectionAccount | None = None + connection_id: str | None = None + owner_kind: Literal["user", "instance", "env"] = "env" + provider: str = "" + name: str = "" + + def __repr__(self) -> str: + return ( + "ResolvedCredential(access_token=SecretStr('**********'), " + f"token_type={self.token_type!r}, expires_at={self.expires_at!r}, " + f"granted_scopes={self.granted_scopes!r}, scopes_verified={self.scopes_verified!r}, " + f"account={self.account!r}, connection_id={self.connection_id!r}, " + f"owner_kind={self.owner_kind!r}, provider={self.provider!r}, name={self.name!r})" + ) + + def __reduce__(self): + msg = "ResolvedCredential objects cannot be serialized" + raise TypeError(msg) + + +@dataclass(frozen=True, slots=True) +class ConnectionResolutionRequest: + """All non-secret context a host needs to resolve one connection.""" + + ref: ConnectionRef + principal: ExecutionPrincipal + required_scopes: frozenset[str] = frozenset() + component_id: str | None = None + flow_id: str | None = None + run_id: str | None = None + + +@dataclass(frozen=True, slots=True) +class ConnectionStatus: + """Credential-free connection status suitable for pickers and health views.""" + + ref: ConnectionRef + status: Literal["ready", "expired", "missing", "scope_missing", "unavailable"] + granted_scopes: frozenset[str] = frozenset() + account: ConnectionAccount | None = None + + +class CredentialLease: + """In-process, single-flight lease for a resolver-provided credential.""" + + def __init__( + self, + resolver: ConnectionResolverProtocol, + request: ConnectionResolutionRequest, + *, + now: Callable[[], datetime] | None = None, + ) -> None: + self._resolver = resolver + self._request = request + self._credential: ResolvedCredential | None = None + self._lock = asyncio.Lock() + self._now = now or (lambda: datetime.now(timezone.utc)) + self._reactive_refresh_completed = False + + @property + def ref(self) -> ConnectionRef: + """Return the non-secret reference represented by this lease.""" + return self._request.ref + + @property + def credential(self) -> ResolvedCredential | None: + """Return the currently cached credential without resolving it.""" + return self._credential + + def _expires_soon(self, credential: ResolvedCredential) -> bool: + expires_at = credential.expires_at + if expires_at is None: + return False + if expires_at.tzinfo is None: + expires_at = expires_at.replace(tzinfo=timezone.utc) + return expires_at - self._now() < _EXPIRY_MARGIN + + async def get_credential(self) -> ResolvedCredential: + """Resolve once, refreshing under one lock when the cached token nears expiry.""" + credential = self._credential + if credential is not None and not self._expires_soon(credential): + return credential + async with self._lock: + credential = self._credential + if credential is None or self._expires_soon(credential): + credential = await self._resolver.resolve(self._request) + self._credential = credential + return credential + + async def get_token(self) -> str: + """Return the access token for immediate use at the provider boundary.""" + credential = await self.get_credential() + return credential.access_token.get_secret_value() + + async def get_token_after_auth_error(self, error: AuthExpiredError) -> str: + """Re-resolve once after a provider rejects a no-expiry or stale token.""" + from lfx.integrations.errors import AuthExpiredError + + if not isinstance(error, AuthExpiredError): + msg = "error must be an AuthExpiredError" + raise TypeError(msg) + async with self._lock: + if self._reactive_refresh_completed: + raise error + self._reactive_refresh_completed = True + self._credential = await self._resolver.resolve(self._request) + credential = self._credential + return credential.access_token.get_secret_value() diff --git a/src/lfx/src/lfx/integrations/telemetry.py b/src/lfx/src/lfx/integrations/telemetry.py new file mode 100644 index 000000000000..f0ae9ece0df3 --- /dev/null +++ b/src/lfx/src/lfx/integrations/telemetry.py @@ -0,0 +1,93 @@ +"""Low-cardinality telemetry boundary for integration actions.""" + +from __future__ import annotations + +import contextlib +import time +from contextlib import asynccontextmanager +from typing import TYPE_CHECKING + +from lfx.integrations.errors import INTEGRATION_ERROR_CODES, IntegrationError, normalize_integration_error +from lfx.observability import outbound_call_span +from lfx.services.schema import ServiceType +from lfx.services.telemetry.schema import IntegrationActionPayload + +if TYPE_CHECKING: + from collections.abc import AsyncIterator + + from lfx.custom.custom_component.component import Component + + +async def _emit(payload: IntegrationActionPayload) -> None: + """Best-effort enqueue without creating a telemetry service as a side effect.""" + try: + from lfx.services.manager import get_service_manager + + service = get_service_manager().services.get(ServiceType.TELEMETRY_SERVICE) + if service is not None: + await service.log_integration_action(payload) + except Exception: # noqa: BLE001 - telemetry must never take down an action + return + + +@asynccontextmanager +async def integration_action( + component: Component, + *, + provider: str, + capability: str, + owner_kind: str, +) -> AsyncIterator[None]: + """Measure, classify, and safely trace one provider action.""" + started = time.monotonic() + principal = getattr(getattr(component, "graph", None), "execution_principal", None) + principal_kind = getattr(principal, "kind", "unknown") + error_code: str | None = None + success = False + attributes = { + "integration.provider": provider, + "integration.capability": capability, + "integration.owner_kind": owner_kind, + "integration.principal_kind": principal_kind, + } + try: + with outbound_call_span("integration.action", attributes) as span: + try: + yield + success = True + except IntegrationError as exc: + error_code = exc.code + span.set_attribute("integration.error_code", error_code) + span.record_error(error_code) + raise + except Exception as exc: + normalized = normalize_integration_error(exc, provider=provider) + error_code = normalized.code + span.set_attribute("integration.error_code", error_code) + span.record_error(error_code) + raise normalized from exc + else: + span.set_attribute("integration.error_code", "none") + finally: + elapsed_ms = max(0, int((time.monotonic() - started) * 1000)) + payload = IntegrationActionPayload( + provider=provider, + capability=capability, + ms=elapsed_ms, + success=success, + error_code=error_code if error_code in INTEGRATION_ERROR_CODES else ("other" if error_code else None), + owner_kind=owner_kind, + principal_kind=principal_kind, + ) + await _emit(payload) + with contextlib.suppress(Exception): + component.log( + { + "provider": provider, + "capability": capability, + "success": success, + "error_code": payload.error_code, + "ms": elapsed_ms, + }, + name="Integration action", + ) diff --git a/src/lfx/src/lfx/io/__init__.py b/src/lfx/src/lfx/io/__init__.py index 632cf77e5e24..c36999aa0135 100644 --- a/src/lfx/src/lfx/io/__init__.py +++ b/src/lfx/src/lfx/io/__init__.py @@ -1,6 +1,7 @@ from lfx.inputs import ( BoolInput, CodeInput, + ConnectionRefInput, DataDisplayInput, DataFrameInput, DataInput, @@ -36,6 +37,7 @@ __all__ = [ "BoolInput", "CodeInput", + "ConnectionRefInput", "DBProviderInput", "DataDisplayInput", "DataFrameInput", diff --git a/src/lfx/src/lfx/io/schema.py b/src/lfx/src/lfx/io/schema.py index acbd09cbc32d..6bd58b3701ff 100644 --- a/src/lfx/src/lfx/io/schema.py +++ b/src/lfx/src/lfx/io/schema.py @@ -54,6 +54,7 @@ FieldTypes.ACTION_PICKER.value: list, FieldTypes.DURATION.value: dict, FieldTypes.CONNECTION.value: str, + FieldTypes.CONNECTION_REF.value: str, FieldTypes.AUTH.value: dict, FieldTypes.FILE.value: str, FieldTypes.PROMPT.value: str, @@ -328,6 +329,8 @@ def create_input_schema(inputs: list["InputTypes"]) -> type[BaseModel]: raise TypeError(msg) fields = {} for input_model in inputs: + if input_model.field_type == FieldTypes.CONNECTION_REF: + continue # Create a Pydantic Field for each input field field_type = input_model.field_type if isinstance(field_type, FieldTypes): @@ -374,6 +377,8 @@ def create_input_schema_from_dict(inputs: list[dotdict], param_key: str | None = raise TypeError(msg) fields = {} for input_model in inputs: + if input_model.type == FieldTypes.CONNECTION_REF.value: + continue # Create a Pydantic Field for each input field try: field_type = _serialized_field_type_to_type[input_model.type] diff --git a/src/lfx/src/lfx/run/_defaults.py b/src/lfx/src/lfx/run/_defaults.py index 06bf730e46e1..b9f48174ad8c 100644 --- a/src/lfx/src/lfx/run/_defaults.py +++ b/src/lfx/src/lfx/run/_defaults.py @@ -24,6 +24,7 @@ from typing import TYPE_CHECKING from lfx.log.logger import logger +from lfx.services.authorization.base import ExecutionPrincipal from lfx.services.deps import get_settings_service if TYPE_CHECKING: @@ -89,6 +90,14 @@ def apply_run_defaults( else: # Caller-supplied None plus an existing graph.user_id: preserve the existing. user_id = graph.user_id + graph.execution_principal = ExecutionPrincipal( + kind="headless_operator", + user_id=str(user_id), + actor_id=str(user_id), + family="lfx_headless", + interactive=True, + actor_label=str(user_id), + ) if not session_id: session_id = uuid.uuid4().hex diff --git a/src/lfx/src/lfx/run/base.py b/src/lfx/src/lfx/run/base.py index ce0cbffa907b..f5ffa0962626 100644 --- a/src/lfx/src/lfx/run/base.py +++ b/src/lfx/src/lfx/run/base.py @@ -8,17 +8,26 @@ from pathlib import Path from typing import TYPE_CHECKING +from lfx.cli.runtime_variables import build_request_variables_from_global_vars from lfx.cli.script_loader import ( extract_structured_result, extract_text_from_result, find_graph_variable, load_graph_from_script, ) -from lfx.cli.validation import validate_global_variables_for_env +from lfx.cli.validation import validate_connection_refs_for_env, validate_global_variables_for_env from lfx.execution import aget_default_coordinator from lfx.log.logger import logger from lfx.run._defaults import apply_run_defaults, resolve_fallback_to_env_vars, validate_provided_id from lfx.schema.schema import InputValueRequest +from lfx.services.variable.request_scope import ( + activate_no_env_fallback, + activate_request_variables, + get_active_request_variables, + is_env_fallback_disabled, + reset_no_env_fallback, + reset_request_variables, +) from lfx.utils.flow_envelope import split_flow_envelope if TYPE_CHECKING: @@ -458,6 +467,11 @@ async def run_flow( logger.debug(f"Validation error: {error}") output_error(error_details, verbose=verbose) raise RunError(error_details, None) + connection_errors = validate_connection_refs_for_env(graph) + if connection_errors: + error = connection_errors[0] + output_error(str(error), verbose=verbose, exception=error) + raise RunError(str(error), error) logger.info("Global variable validation passed") else: logger.info("Global variable validation skipped") @@ -494,6 +508,13 @@ async def run_flow( execution_step_start = execution_start_time if timing else None result_count = 0 + # Match serve execution: resolver lookups must see this graph's injection + # channel and environment policy, including during human-input execution. + scope_vars = build_request_variables_from_global_vars(graph.context.get("request_variables")) + scope_token = activate_request_variables(scope_vars or get_active_request_variables()) + no_env_token = activate_no_env_fallback( + disabled=bool(graph.context.get("no_env_fallback")) or is_env_fallback_disabled() + ) try: sys.stdout = captured_stdout # Don't capture stderr at high verbosity levels to avoid duplication with direct logging @@ -637,6 +658,8 @@ async def run_flow( finally: sys.stdout = original_stdout sys.stderr = original_stderr + reset_no_env_fallback(no_env_token) + reset_request_variables(scope_token) execution_end_time = time.monotonic() if timing else None diff --git a/src/lfx/src/lfx/services/__init__.py b/src/lfx/src/lfx/services/__init__.py index c258165e05e2..9965fc6b012d 100644 --- a/src/lfx/src/lfx/services/__init__.py +++ b/src/lfx/src/lfx/services/__init__.py @@ -1,11 +1,12 @@ """LFX services module - pluggable service architecture for dependency injection.""" from .adapters.registry import register_adapter, teardown_all_adapter_registries -from .deps import get_deployment_adapter +from .deps import get_connection_resolver, get_deployment_adapter from .interfaces import ( AuthServiceProtocol, CacheServiceProtocol, ChatServiceProtocol, + ConnectionResolverProtocol, DatabaseServiceProtocol, DeploymentServiceProtocol, SettingsServiceProtocol, @@ -22,6 +23,7 @@ "AuthServiceProtocol", "CacheServiceProtocol", "ChatServiceProtocol", + "ConnectionResolverProtocol", "DatabaseServiceProtocol", "DeploymentServiceProtocol", "MCPComposerService", @@ -32,6 +34,7 @@ "StorageServiceProtocol", "TracingServiceProtocol", "VariableServiceProtocol", + "get_connection_resolver", "get_deployment_adapter", "register_adapter", "register_service", diff --git a/src/lfx/src/lfx/services/authorization/__init__.py b/src/lfx/src/lfx/services/authorization/__init__.py index a53d6539720f..72d58aec6246 100644 --- a/src/lfx/src/lfx/services/authorization/__init__.py +++ b/src/lfx/src/lfx/services/authorization/__init__.py @@ -12,6 +12,7 @@ DirectoryMembershipClaimState, DirectoryMembershipIngestResult, DirectoryMembershipSnapshot, + ExecutionPrincipal, PublicAuthorizationRequest, PublicResourceAction, ResourceVisibilityScope, @@ -33,6 +34,7 @@ "DirectoryMembershipClaimState", "DirectoryMembershipIngestResult", "DirectoryMembershipSnapshot", + "ExecutionPrincipal", "PublicAuthorizationRequest", "PublicResourceAction", "ResourceVisibilityScope", diff --git a/src/lfx/src/lfx/services/authorization/base.py b/src/lfx/src/lfx/services/authorization/base.py index e0d86b74bb0f..fbcbfc574c1d 100644 --- a/src/lfx/src/lfx/services/authorization/base.py +++ b/src/lfx/src/lfx/services/authorization/base.py @@ -42,6 +42,15 @@ class AuthzContext(TypedDict, total=False): PUBLIC_ANONYMOUS_ACTOR_ID = uuid5(NAMESPACE_URL, "urn:langflow:principal:anonymous-public") AdministrationResource = Literal["user", "team", "role"] +ExecutionPrincipalKind = Literal[ + "actor", + "flow_owner", + "deployment_owner", + "job_owner", + "anonymous_public", + "headless_operator", + "unknown", +] class PublicResourceAction(str, Enum): @@ -70,6 +79,24 @@ def public_anonymous(cls) -> AuthorizationPrincipal: return cls(actor_type="anonymous_public", actor_id=PUBLIC_ANONYMOUS_ACTOR_ID) +@dataclass(frozen=True, slots=True) +class ExecutionPrincipal: + """Identity and route family used for dependency credential resolution.""" + + kind: ExecutionPrincipalKind + user_id: str | None = None + actor_id: str | None = None + family: str | None = None + interactive: bool = False + end_user_id: str | None = None + actor_label: str | None = None + + @classmethod + def unknown(cls) -> ExecutionPrincipal: + """Return the fail-closed principal for unstamped execution paths.""" + return cls(kind="unknown") + + @dataclass(frozen=True, slots=True) class PublicAuthorizationRequest: """Plugin-neutral anonymous resource decision. diff --git a/src/lfx/src/lfx/services/connection/__init__.py b/src/lfx/src/lfx/services/connection/__init__.py new file mode 100644 index 000000000000..78869a6389c7 --- /dev/null +++ b/src/lfx/src/lfx/services/connection/__init__.py @@ -0,0 +1,11 @@ +"""Connection resolver services.""" + +from lfx.services.connection.base import BaseConnectionResolverService, ConnectionAccessPolicy +from lfx.services.connection.env_resolver import EnvConnectionResolver, RequestScopedConnectionResolver + +__all__ = [ + "BaseConnectionResolverService", + "ConnectionAccessPolicy", + "EnvConnectionResolver", + "RequestScopedConnectionResolver", +] diff --git a/src/lfx/src/lfx/services/connection/base.py b/src/lfx/src/lfx/services/connection/base.py new file mode 100644 index 000000000000..7d9a34b56a9a --- /dev/null +++ b/src/lfx/src/lfx/services/connection/base.py @@ -0,0 +1,140 @@ +"""Host-pluggable connection resolver contract.""" + +from __future__ import annotations + +import abc +from typing import TYPE_CHECKING, Literal, final + +from pydantic import BaseModel, ConfigDict, StrictStr + +from lfx.integrations.capabilities import ScopeSet +from lfx.integrations.errors import ConnectionNotAuthorizedError, IntegrationError, ScopeMissingError +from lfx.services.base import Service +from lfx.services.schema import ServiceType + +if TYPE_CHECKING: + from lfx.integrations.models import ( + ConnectionRef, + ConnectionResolutionRequest, + ConnectionStatus, + ResolvedCredential, + ) + from lfx.services.authorization.base import ExecutionPrincipal + + +class ConnectionAccessPolicy(BaseModel): + """Host-owned metadata, loaded without decrypting or refreshing credentials. + + This policy must describe the same connection passed to ``_resolve``. Share + decisions come from host authorization, never flow JSON or component input. + """ + + model_config = ConfigDict(extra="forbid", frozen=True, strict=True) + + owner_kind: Literal["user", "instance", "env"] + connection_owner_id: StrictStr | None = None + connection_id: StrictStr | None = None + allow_non_interactive: bool = False + explicit_share_authorized: bool = False + + +class BaseConnectionResolverService(Service, abc.ABC): + """Resolve portable connection handles inside the current host boundary.""" + + name = ServiceType.CONNECTION_RESOLVER_SERVICE.value + + def __init_subclass__(cls) -> None: + super().__init_subclass__() + if cls.resolve is not BaseConnectionResolverService.resolve: + msg = "Connection resolvers must implement _get_access_policy and _resolve; resolve cannot be overridden" + raise TypeError(msg) + + @final + async def resolve(self, request: ConnectionResolutionRequest) -> ResolvedCredential: + """Enforce the portable floor before invoking the host's credential hook.""" + if request.principal.kind in {"anonymous_public", "unknown"}: + raise ConnectionNotAuthorizedError(provider=request.ref.provider) + policy = await self._get_access_policy(request) + if not isinstance(policy, ConnectionAccessPolicy): + raise ConnectionNotAuthorizedError(provider=request.ref.provider) + denial = BaseConnectionResolverService.authorize_principal( + self, + request, + connection_owner_id=policy.connection_owner_id, + owner_kind=policy.owner_kind, + allow_non_interactive=policy.allow_non_interactive, + explicit_share_authorized=policy.explicit_share_authorized, + ) + if denial is not None: + raise denial + credential = await self._resolve(request, policy) + if request.required_scopes and not credential.scopes_verified: + raise ScopeMissingError(request.required_scopes, provider=request.ref.provider, scopes_verified=False) + missing = ScopeSet.missing( + provider=request.ref.provider, required=request.required_scopes, granted=credential.granted_scopes + ) + if missing: + raise ScopeMissingError(frozenset(missing), provider=request.ref.provider) + return credential + + @abc.abstractmethod + async def _get_access_policy(self, request: ConnectionResolutionRequest) -> ConnectionAccessPolicy: + """Load ownership/opt-in metadata and verify any explicit share, without reading secrets.""" + + @abc.abstractmethod + async def _resolve( + self, request: ConnectionResolutionRequest, policy: ConnectionAccessPolicy + ) -> ResolvedCredential: + """Read/refresh only the connection identified by the authorized policy. + + Hosts must keep policy and credential lookup consistent, using the same + connection id and checking for ownership/policy changes during resolution. + """ + + async def describe( + self, + ref: ConnectionRef, + principal: ExecutionPrincipal, + ) -> ConnectionStatus | None: + """Return credential-free status when the host supports discovery.""" + _ = (ref, principal) + return None + + def authorize_principal( + self, + request: ConnectionResolutionRequest, + *, + connection_owner_id: str | None, + owner_kind: Literal["user", "instance", "env"], + allow_non_interactive: bool, + explicit_share_authorized: bool = False, + ) -> IntegrationError | None: + """Apply the portable deny floor, including a host-verified share decision. + + Only a host may set ``explicit_share_authorized``, after checking the + actor's connection:execute grant and the route family's share policy. + It must never come from flow JSON or component input. A share can satisfy + an actor's owner mismatch; it cannot override any other deny below. + """ + principal = request.principal + if owner_kind == "env": + return ( + None + if principal.kind == "headless_operator" + else ConnectionNotAuthorizedError(provider=request.ref.provider) + ) + if principal.kind in {"anonymous_public", "unknown"}: + return ConnectionNotAuthorizedError(provider=request.ref.provider) + if owner_kind == "user": + if connection_owner_id is None or principal.user_id is None: + return ConnectionNotAuthorizedError(provider=request.ref.provider) + if not principal.interactive and not allow_non_interactive: + return ConnectionNotAuthorizedError(provider=request.ref.provider) + if str(principal.user_id) != str(connection_owner_id) and not ( + principal.kind == "actor" and explicit_share_authorized + ): + return ConnectionNotAuthorizedError(provider=request.ref.provider) + return None + + async def teardown(self) -> None: + """Resolvers own no resources by default.""" diff --git a/src/lfx/src/lfx/services/connection/env_resolver.py b/src/lfx/src/lfx/services/connection/env_resolver.py new file mode 100644 index 000000000000..6b4525089975 --- /dev/null +++ b/src/lfx/src/lfx/services/connection/env_resolver.py @@ -0,0 +1,160 @@ +"""Environment and request-scope resolver for headless lfx runtimes.""" + +from __future__ import annotations + +import json +from datetime import datetime, timezone +from typing import TYPE_CHECKING, Any + +from pydantic import SecretStr + +from lfx.integrations.errors import AuthExpiredError, ConnectionUnresolvedError +from lfx.integrations.models import ( + ConnectionAccount, + ConnectionResolutionRequest, + ResolvedCredential, +) +from lfx.services.connection.base import BaseConnectionResolverService, ConnectionAccessPolicy +from lfx.services.variable.request_scope import is_env_fallback_disabled + +if TYPE_CHECKING: + from lfx.integrations.errors import ConnectionUnresolvedReason + + +class _InvalidCredentialError(ValueError): + """Carry only a fixed reason across the private parser boundary.""" + + def __init__(self, *, reason: ConnectionUnresolvedReason) -> None: + self.reason = reason + super().__init__(reason) + + +def _parse_expiry(value: Any) -> datetime | None: + """Interpret optional timestamps as UTC, rejecting invalid wire values.""" + if value is None: + return None + if isinstance(value, int | float) and not isinstance(value, bool): + return datetime.fromtimestamp(value, tz=timezone.utc) + if not isinstance(value, str): + msg = "expires_at must be an ISO-8601 string or Unix timestamp" + raise TypeError(msg) + normalized = value[:-1] + "+00:00" if value.endswith("Z") else value + parsed = datetime.fromisoformat(normalized) + return parsed.replace(tzinfo=timezone.utc) if parsed.tzinfo is None else parsed + + +def _parse_wire_value(raw: str, request: ConnectionResolutionRequest) -> ResolvedCredential: + """Keep raw wire values and validation exceptions out of public error chains.""" + reason: ConnectionUnresolvedReason = "invalid-credential" + try: + return _parse_credential(raw, request) + except _InvalidCredentialError as exc: + reason = exc.reason + except (ValueError, TypeError, OverflowError, OSError): + pass + # Raise outside the handler so even __context__ cannot retain a raw value. + raise ConnectionUnresolvedError( + request.ref.to_handle(), env_key=request.ref.env_key(), provider=request.ref.provider, reason=reason + ) + + +def _parse_credential(raw: str, request: ConnectionResolutionRequest) -> ResolvedCredential: + """Validate a token or credential JSON inside the sanitized parser boundary.""" + if not raw: + raise _InvalidCredentialError(reason="invalid-access-token") + if not raw.lstrip().startswith("{"): + return ResolvedCredential( + access_token=SecretStr(raw), + provider=request.ref.provider, + name=request.ref.name, + owner_kind="env", + ) + try: + payload = json.loads(raw) + except json.JSONDecodeError: + raise _InvalidCredentialError(reason="malformed-json") from None + if not isinstance(payload, dict): + raise _InvalidCredentialError(reason="malformed-json") + forbidden = {"refresh_token", "client_secret", "password"} & payload.keys() + if forbidden: + raise _InvalidCredentialError(reason="long-lived-secret") + allowed = {"access_token", "token_type", "expires_at", "scopes", "account"} + unknown = set(payload) - allowed + if unknown: + raise _InvalidCredentialError(reason="unsupported-fields") + access_token = payload.get("access_token") + if not isinstance(access_token, str) or not access_token: + raise _InvalidCredentialError(reason="invalid-access-token") + scopes = payload.get("scopes", []) + if not isinstance(scopes, list) or any(not isinstance(scope, str) or not scope for scope in scopes): + raise _InvalidCredentialError(reason="invalid-scopes") + token_type = payload.get("token_type", "Bearer") + if not isinstance(token_type, str) or not token_type: + raise _InvalidCredentialError(reason="invalid-token-type") + account_payload = payload.get("account") + try: + account = ConnectionAccount.model_validate(account_payload) if account_payload is not None else None + except (ValueError, TypeError): + raise _InvalidCredentialError(reason="invalid-account") from None + try: + expires_at = _parse_expiry(payload.get("expires_at")) + except (ValueError, TypeError, OverflowError, OSError): + raise _InvalidCredentialError(reason="invalid-expiry") from None + return ResolvedCredential( + access_token=SecretStr(access_token), + token_type=token_type, + expires_at=expires_at, + granted_scopes=frozenset(scopes), + scopes_verified="scopes" in payload, + account=account, + provider=request.ref.provider, + name=request.ref.name, + owner_kind="env", + ) + + +class EnvConnectionResolver(BaseConnectionResolverService): + """Resolve credentials through the existing variable/request-scope service.""" + + def __init__(self) -> None: + super().__init__() + self._fallback_variable_service = None + self.set_ready() + + async def _get_access_policy(self, request: ConnectionResolutionRequest) -> ConnectionAccessPolicy: + _ = request + return ConnectionAccessPolicy(owner_kind="env", allow_non_interactive=True) + + async def _resolve( + self, request: ConnectionResolutionRequest, policy: ConnectionAccessPolicy + ) -> ResolvedCredential: + """Resolve and validate the bare-token or JSON headless wire format.""" + _ = policy + from lfx.services.deps import get_variable_service + + variable_service = get_variable_service() + if variable_service is None: + from lfx.services.variable.service import VariableService + + if self._fallback_variable_service is None: + self._fallback_variable_service = VariableService() + variable_service = self._fallback_variable_service + raw = await variable_service.get_variable(request.ref.env_key()) + if raw is None: + raise ConnectionUnresolvedError( + request.ref.to_handle(), + env_key=request.ref.env_key(), + provider=request.ref.provider, + reason="env-fallback-disabled" if is_env_fallback_disabled() else "missing", + ) + credential = _parse_wire_value(str(raw), request) + if credential.expires_at is not None and credential.expires_at <= datetime.now(timezone.utc): + raise AuthExpiredError(provider=request.ref.provider) + return credential + + async def teardown(self) -> None: + if self._fallback_variable_service is not None: + await self._fallback_variable_service.teardown() + + +RequestScopedConnectionResolver = EnvConnectionResolver diff --git a/src/lfx/src/lfx/services/deps.py b/src/lfx/src/lfx/services/deps.py index 78a63d9f71c7..4731788e8154 100644 --- a/src/lfx/src/lfx/services/deps.py +++ b/src/lfx/src/lfx/services/deps.py @@ -24,6 +24,7 @@ AuthServiceProtocol, CacheServiceProtocol, ChatServiceProtocol, + ConnectionResolverProtocol, DatabaseServiceProtocol, DeploymentServiceProtocol, SettingsServiceProtocol, @@ -175,6 +176,56 @@ def get_variable_service() -> VariableServiceProtocol | None: return get_service(ServiceType.VARIABLE_SERVICE) +_connection_resolver_fallback_lock = threading.Lock() + + +def get_connection_resolver() -> ConnectionResolverProtocol: + """Return the configured resolver or the stable headless environment fallback. + + Only a genuinely absent plugin selects the fallback. Import, construction, + type, and readiness failures remain operator-visible so a broken connection + plugin cannot silently change credential sources. + """ + from lfx.services.connection.base import BaseConnectionResolverService + from lfx.services.connection.env_resolver import EnvConnectionResolver + from lfx.services.manager import NoFactoryRegisteredError, get_service_manager + + service_manager = get_service_manager() + cached = service_manager.services.get(ServiceType.CONNECTION_RESOLVER_SERVICE) + if cached is not None: + if not isinstance(cached, BaseConnectionResolverService) or not cached.ready: + msg = "A configured connection_resolver_service must be valid and ready" + raise RuntimeError(msg) + return cast("ConnectionResolverProtocol", cached) + + try: + service = service_manager.get(ServiceType.CONNECTION_RESOLVER_SERVICE) + except NoFactoryRegisteredError: + if ( + ServiceType.CONNECTION_RESOLVER_SERVICE in service_manager.service_classes + or ServiceType.CONNECTION_RESOLVER_SERVICE in service_manager.factories + ): + raise # A configured resolver can fail because one of its dependencies is absent. + service = None + + if service is not None: + if not isinstance(service, BaseConnectionResolverService) or not service.ready: + msg = "A configured connection_resolver_service must be valid and ready" + raise RuntimeError(msg) + return cast("ConnectionResolverProtocol", service) + + with _connection_resolver_fallback_lock: + cached = service_manager.services.get(ServiceType.CONNECTION_RESOLVER_SERVICE) + if cached is not None: + if not isinstance(cached, BaseConnectionResolverService) or not cached.ready: + msg = "A configured connection_resolver_service must be valid and ready" + raise RuntimeError(msg) + return cast("ConnectionResolverProtocol", cached) + if service_manager.connection_resolver_fallback is None: + service_manager.connection_resolver_fallback = EnvConnectionResolver() + return service_manager.connection_resolver_fallback + + def get_shared_component_cache_service() -> CacheServiceProtocol | None: """Retrieves the shared component cache service instance.""" from lfx.services.shared_component_cache.factory import SharedComponentCacheServiceFactory diff --git a/src/lfx/src/lfx/services/interfaces.py b/src/lfx/src/lfx/services/interfaces.py index 5cbf8ebaf888..43f53b8812b1 100644 --- a/src/lfx/src/lfx/services/interfaces.py +++ b/src/lfx/src/lfx/services/interfaces.py @@ -11,6 +11,12 @@ from sqlalchemy.ext.asyncio import AsyncSession + from lfx.integrations.models import ( + ConnectionRef, + ConnectionResolutionRequest, + ConnectionStatus, + ResolvedCredential, + ) from lfx.services.adapters.deployment.schema import ( ConfigListParams, ConfigListResult, @@ -37,6 +43,7 @@ VerifyCredentials, VerifyCredentialsResult, ) + from lfx.services.authorization.base import ExecutionPrincipal from lfx.services.settings.base import Settings @@ -153,6 +160,25 @@ async def get_all_decrypted_variables(self, user_id: Any, session: Any) -> dict[ ... +@runtime_checkable +class ConnectionResolverProtocol(Protocol): + """Portable surface implemented by host connection resolvers.""" + + @abstractmethod + async def resolve(self, request: ConnectionResolutionRequest) -> ResolvedCredential: + """Resolve a non-secret reference for the current execution principal.""" + ... + + @abstractmethod + async def describe( + self, + ref: ConnectionRef, + principal: ExecutionPrincipal, + ) -> ConnectionStatus | None: + """Return credential-free connection status when supported.""" + ... + + class CacheServiceProtocol(Protocol): """Protocol for cache service.""" diff --git a/src/lfx/src/lfx/services/manager.py b/src/lfx/src/lfx/services/manager.py index e1291a34bbca..2c2700f36e88 100644 --- a/src/lfx/src/lfx/services/manager.py +++ b/src/lfx/src/lfx/services/manager.py @@ -29,6 +29,7 @@ if TYPE_CHECKING: from lfx.services.base import Service + from lfx.services.connection.base import BaseConnectionResolverService from lfx.services.factory import ServiceFactory @@ -52,6 +53,7 @@ def __init__(self) -> None: self.keyed_lock = KeyedMemoryLockManager() self.factory_registered = False self._plugins_discovered = False + self.connection_resolver_fallback: BaseConnectionResolverService | None = None # Always register settings service from lfx.services.settings.factory import SettingsServiceFactory @@ -291,7 +293,7 @@ async def teardown(self, *, raise_on_error: bool = False) -> None: first error after the table is cleared. Default False logs failures only. """ errors: list[tuple[str, Exception]] = [] - for service in list(self.services.values()): + for service in [*self.services.values(), self.connection_resolver_fallback]: if service is None: continue # Registered services are duck-typed: the in-memory caches and the Noop @@ -321,6 +323,7 @@ async def teardown(self, *, raise_on_error: bool = False) -> None: errors.append(("adapter_registries", exc)) self.services = {} + self.connection_resolver_fallback = None self.factories = {} # ``teardown`` empties the factory registry, so the "registered" flag has # to drop too: get_service() re-registers factories only when @@ -429,6 +432,12 @@ def _discover_from_entry_points(self) -> None: expected_bases[ServiceType.AUTHORIZATION_SERVICE] = BaseAuthorizationService except Exception as exc: # noqa: BLE001 — optional import, validation just skipped logger.debug(f"BaseAuthorizationService unavailable; entry-point validation skipped: {exc}") + try: + from lfx.services.connection.base import BaseConnectionResolverService + + expected_bases[ServiceType.CONNECTION_RESOLVER_SERVICE] = BaseConnectionResolverService + except Exception as exc: # noqa: BLE001 — optional import, validation just skipped + logger.debug(f"BaseConnectionResolverService unavailable; entry-point validation skipped: {exc}") try: from lfx.services.catalog_policy.base import BaseCatalogPolicyService @@ -457,6 +466,12 @@ def _discover_from_entry_points(self) -> None: if expected_base is not None and not ( isinstance(service_class, type) and issubclass(service_class, expected_base) ): + if service_type == ServiceType.CONNECTION_RESOLVER_SERVICE: + msg = ( + "Connection resolver entry point must subclass " + f"{expected_base.__name__}; refusing to use the environment fallback" + ) + raise RuntimeError(msg) logger.warning( f"Entry point {ep.name} resolved to {service_class!r}, " f"which is not a subclass of {expected_base.__name__}. " @@ -466,12 +481,22 @@ def _discover_from_entry_points(self) -> None: continue self.register_service_class(service_type, service_class, override=False) logger.debug(f"Loaded service from entry point: {ep.name}") + except RuntimeError as exc: + if ep.name == ServiceType.CONNECTION_RESOLVER_SERVICE.value: + raise + logger.warning(f"Error loading entry point {ep.name}: {exc}") except (ValueError, AttributeError) as exc: + if ep.name == ServiceType.CONNECTION_RESOLVER_SERVICE.value: + msg = "Connection resolver entry point failed to load; refusing to use the environment fallback" + raise RuntimeError(msg) from exc logger.warning(f"Failed to load entry point {ep.name}: {exc}") - except Exception as exc: # noqa: BLE001 + except Exception as exc: # Authz plugin failures are operator-visible — silent # degradation to the OSS pass-through is exactly the kind # of behavior change we want noisy. + if ep.name == ServiceType.CONNECTION_RESOLVER_SERVICE.value: + msg = "Connection resolver entry point failed to load; refusing to use the environment fallback" + raise RuntimeError(msg) from exc logger.warning(f"Error loading entry point {ep.name}: {exc}") def _discover_from_config(self, config_dir: Path) -> None: @@ -518,6 +543,12 @@ def _register_service_from_path(self, service_key: str, service_path: str) -> No object_key=service_key, ) if service_class is None: + if service_type == ServiceType.CONNECTION_RESOLVER_SERVICE: + msg = ( + "Configured connection resolver service could not be loaded; " + "refusing to use the environment fallback" + ) + raise RuntimeError(msg) if service_type == ServiceType.MODEL_PROVIDER_POLICY_SERVICE: msg = ( "Configured model provider policy service could not be loaded; " @@ -532,6 +563,12 @@ def _register_service_from_path(self, service_key: str, service_path: str) -> No raise RuntimeError(msg) return + if service_type == ServiceType.CONNECTION_RESOLVER_SERVICE: + from lfx.services.connection.base import BaseConnectionResolverService + + if not isinstance(service_class, type) or not issubclass(service_class, BaseConnectionResolverService): + msg = "Configured connection resolver service must subclass BaseConnectionResolverService" + raise RuntimeError(msg) if service_type == ServiceType.MODEL_PROVIDER_POLICY_SERVICE: from lfx.services.model_provider_policy.base import BaseModelProviderPolicyService diff --git a/src/lfx/src/lfx/services/schema.py b/src/lfx/src/lfx/services/schema.py index aba485aed0a2..1615f0de2e92 100644 --- a/src/lfx/src/lfx/services/schema.py +++ b/src/lfx/src/lfx/services/schema.py @@ -15,6 +15,7 @@ class ServiceType(str, Enum): STORAGE_SERVICE = "storage_service" SETTINGS_SERVICE = "settings_service" VARIABLE_SERVICE = "variable_service" + CONNECTION_RESOLVER_SERVICE = "connection_resolver_service" CACHE_SERVICE = "cache_service" TELEMETRY_SERVICE = "telemetry_service" TRACING_SERVICE = "tracing_service" diff --git a/src/lfx/src/lfx/services/telemetry/__init__.py b/src/lfx/src/lfx/services/telemetry/__init__.py index 583c1e3f9624..b6927ff723cd 100644 --- a/src/lfx/src/lfx/services/telemetry/__init__.py +++ b/src/lfx/src/lfx/services/telemetry/__init__.py @@ -1,6 +1,6 @@ """Telemetry service for lfx package.""" -from .schema import MCPToolPayload +from .schema import IntegrationActionPayload, MCPToolPayload from .service import TelemetryService -__all__ = ["MCPToolPayload", "TelemetryService"] +__all__ = ["IntegrationActionPayload", "MCPToolPayload", "TelemetryService"] diff --git a/src/lfx/src/lfx/services/telemetry/base.py b/src/lfx/src/lfx/services/telemetry/base.py index d5afa250c861..8c92274a82bc 100644 --- a/src/lfx/src/lfx/services/telemetry/base.py +++ b/src/lfx/src/lfx/services/telemetry/base.py @@ -40,6 +40,9 @@ async def log_package_run(self, payload: BaseModel) -> None: payload: Run payload containing run information """ + async def log_integration_action(self, payload: BaseModel) -> None: + """Enqueue an integration event; older host implementations safely omit it.""" + @abstractmethod async def log_package_shutdown(self) -> None: """Log a package shutdown event.""" diff --git a/src/lfx/src/lfx/services/telemetry/schema.py b/src/lfx/src/lfx/services/telemetry/schema.py index 76ff17b3fa5f..a1deeacb4a4b 100644 --- a/src/lfx/src/lfx/services/telemetry/schema.py +++ b/src/lfx/src/lfx/services/telemetry/schema.py @@ -55,3 +55,15 @@ class MCPToolPayload(BasePayload): success: bool ms: int = Field(0, serialization_alias="ms") error: str | None = None + + +class IntegrationActionPayload(BasePayload): + """Low-cardinality integration action event with no connection identifiers.""" + + provider: str + capability: str + ms: int + success: bool + error_code: str | None = Field(None, serialization_alias="errorCode") + owner_kind: str = Field(serialization_alias="ownerKind") + principal_kind: str = Field(serialization_alias="principalKind") diff --git a/src/lfx/src/lfx/services/telemetry/service.py b/src/lfx/src/lfx/services/telemetry/service.py index 35a46182fa3e..29f2a657eb5b 100644 --- a/src/lfx/src/lfx/services/telemetry/service.py +++ b/src/lfx/src/lfx/services/telemetry/service.py @@ -130,6 +130,10 @@ async def send_telemetry_data(self, payload: BaseModel, path: str | None = None) async def log_package_run(self, payload: BaseModel) -> None: await self._enqueue(payload, "run") + async def log_integration_action(self, payload: BaseModel) -> None: + """Queue an integration event without waiting for the telemetry transport.""" + await self._enqueue(payload, "integration_action") + async def log_package_shutdown(self) -> None: elapsed = int((datetime.now(timezone.utc) - self._start_time).total_seconds()) await self._enqueue(ShutdownPayload(time_running=elapsed), "shutdown") diff --git a/src/lfx/src/lfx/utils/constants.py b/src/lfx/src/lfx/utils/constants.py index 0011c59a57e1..55af1c877310 100644 --- a/src/lfx/src/lfx/utils/constants.py +++ b/src/lfx/src/lfx/utils/constants.py @@ -82,6 +82,7 @@ def python_function(text: str) -> str: "duration", "auth", "connect", + "connection_ref", "query", "tools", "mcp", diff --git a/src/lfx/tests/unit/integrations/__init__.py b/src/lfx/tests/unit/integrations/__init__.py new file mode 100644 index 000000000000..7747866861b2 --- /dev/null +++ b/src/lfx/tests/unit/integrations/__init__.py @@ -0,0 +1 @@ +"""Dedicated integration contract tests.""" diff --git a/src/lfx/tests/unit/integrations/test_contracts.py b/src/lfx/tests/unit/integrations/test_contracts.py new file mode 100644 index 000000000000..c1c990cb04fd --- /dev/null +++ b/src/lfx/tests/unit/integrations/test_contracts.py @@ -0,0 +1,179 @@ +from __future__ import annotations + +import json +import pickle +from pathlib import Path +from typing import get_args + +import pytest +from lfx.cli.validation import is_valid_env_var_name +from lfx.extension.manifest import ExtensionManifest +from lfx.inputs.input_mixin import SENSITIVE_FIELD_TYPES, FieldTypes +from lfx.inputs.inputs import ConnectionRefInput, instantiate_input +from lfx.integrations import ( + ConditionalScopeRequirement, + ConnectionRef, + IntegrationCapability, + IntegrationProvider, + OAuthProfile, + ResolvedCredential, + ScopeCondition, + ScopeSet, +) +from lfx.integrations.capabilities import OAuthKind +from lfx.io.schema import create_input_schema, create_input_schema_from_dict +from lfx.schema.dotdict import dotdict +from pydantic import SecretStr, ValidationError + + +@pytest.mark.parametrize("handle", ["google", "google/", "/work", "Google/work", "google/Work", "google/work-a"]) +def test_connection_ref_rejects_malformed_handles(handle: str) -> None: + with pytest.raises((ValueError, ValidationError)): + ConnectionRef.parse(handle) + + +def test_connection_ref_env_key_is_valid_and_collision_free() -> None: + refs = [ConnectionRef.parse(handle) for handle in ("a.b/work", "a-b/work", "a_b/work")] + keys = [ref.env_key() for ref in refs] + + assert keys == [ + "LF_CONNECTION__A_2EB__WORK", + "LF_CONNECTION__A_2DB__WORK", + "LF_CONNECTION__A_5FB__WORK", + ] + assert len(set(keys)) == len(keys) + assert all(is_valid_env_var_name(key) for key in keys) + + +def test_resolved_credential_is_redacted_and_not_picklable() -> None: + credential = ResolvedCredential(access_token=SecretStr("do-not-leak"), provider="google", name="work") + + assert "do-not-leak" not in repr(credential) + with pytest.raises(TypeError, match="cannot be serialized"): + pickle.dumps(credential) + + +def test_connection_ref_input_round_trip_and_tool_exclusion() -> None: + input_model = ConnectionRefInput(name="connection", provider="google", value="google/work") + + assert input_model.field_type == FieldTypes.CONNECTION_REF + assert FieldTypes.CONNECTION_REF in SENSITIVE_FIELD_TYPES + assert input_model.track_in_telemetry is False + assert input_model.load_from_db is False + assert input_model.password is False + serialized = input_model.model_dump(by_alias=True) + serialized.pop("_input_type") + assert isinstance( + instantiate_input("ConnectionRefInput", serialized), + ConnectionRefInput, + ) + assert create_input_schema([input_model]).model_fields == {} + assert create_input_schema_from_dict([dotdict(input_model.to_dict())]).model_fields == {} + + with pytest.raises(ValidationError, match="tool-call"): + ConnectionRefInput(name="connection", provider="google", tool_mode=True) + + +def _provider(provider_id: str = "google") -> IntegrationProvider: + profile = OAuthProfile(id="user", kind="oauth2_authorization_code", identity="user_delegated") + capability = IntegrationCapability( + id="google.drive.read", + display_name="Read Drive", + auth_profile_id="user", + identity="user_delegated", + required_scopes=("drive.read",), + conditional_scopes=( + ConditionalScopeRequirement( + scope="drive.write", + role="optional", + condition=ScopeCondition(kind="input_truthy", input="write"), + ), + ), + risk="read", + component_ref="GoogleDriveComponent", + ) + return IntegrationProvider( + provider_id=provider_id, + display_name="Google", + auth_profiles=(profile,), + capabilities=(capability,), + ) + + +def test_scope_set_activates_conditional_requirements() -> None: + capability = _provider().capabilities[0] + + assert ScopeSet.covers(capability, {"write": False}, {"drive.read"}, provider="google") == frozenset() + assert ScopeSet.covers(capability, {"write": True}, {"drive.read"}, provider="google") == frozenset({"drive.write"}) + + +def test_oauth_profile_kinds_match_discovery_schema() -> None: + schema_path = Path(__file__).parents[5] / "design/dedicated-integrations/schema/capability_matrix.schema.json" + if not schema_path.is_file(): + pytest.skip("The discovery schema is only available in the full repository") + schema = json.loads(schema_path.read_text(encoding="utf-8")) + + assert set(get_args(OAuthKind)) == set(schema["$defs"]["auth_mode"]["enum"]) + + +def test_extension_manifest_accepts_unique_integration_providers() -> None: + manifest_data = { + "id": "lfx-google", + "version": "1.0.0", + "name": "Google", + "lfx": {"compat": ["1"]}, + "bundles": [{"name": "google", "path": "google"}], + "integrations": [_provider().model_dump(mode="json")], + } + + manifest = ExtensionManifest.model_validate(manifest_data) + assert manifest.integrations[0].provider_id == "google" + + +def test_extension_manifest_rejects_duplicate_integration_provider_ids() -> None: + provider = _provider().model_dump(mode="json") + with pytest.raises(ValidationError, match="must be unique"): + ExtensionManifest.model_validate( + { + "id": "lfx-google", + "version": "1.0.0", + "name": "Google", + "lfx": {"compat": ["1"]}, + "bundles": [{"name": "google", "path": "google"}], + "integrations": [provider, provider], + } + ) + + +def test_integration_provider_rejects_profile_identity_mismatch() -> None: + provider = _provider().model_dump(mode="json") + provider["capabilities"][0]["identity"] = "bot" + + with pytest.raises(ValidationError, match="identity does not match"): + IntegrationProvider.model_validate(provider) + + +@pytest.mark.parametrize( + ("provider", "required", "granted"), + [ + ("google", "https://www.googleapis.com/auth/drive.readonly", "drive.readonly"), + ("google_workspace", "https://www.googleapis.com/auth/drive.readonly", "drive.readonly"), + ("microsoft", "https://graph.microsoft.com/Mail.Read", "mail.read"), + ], +) +def test_scope_coverage_uses_explicit_provider_for_unqualified_capability_ids( + provider: str, required: str, granted: str +) -> None: + capability = _provider().capabilities[0].model_copy(update={"id": "read", "required_scopes": (required,)}) + + assert ScopeSet.covers(capability, {}, {granted}, provider=provider) == frozenset() + assert ScopeSet.missing(provider=provider, required={required}, granted={granted}) == frozenset() + assert ScopeSet.missing(provider=provider, required={required}, granted=set()) == frozenset({required}) + + +def test_scope_coverage_requires_provider_and_preserves_distinct_slack_scopes() -> None: + with pytest.raises(TypeError, match="provider"): + ScopeSet.covers(_provider().capabilities[0], {}, set()) + assert ScopeSet.missing(provider="slack", required={"chat:write"}, granted={"chat:write:user"}) == frozenset( + {"chat:write"} + ) diff --git a/src/lfx/tests/unit/integrations/test_errors.py b/src/lfx/tests/unit/integrations/test_errors.py new file mode 100644 index 000000000000..200c263a42ed --- /dev/null +++ b/src/lfx/tests/unit/integrations/test_errors.py @@ -0,0 +1,123 @@ +from __future__ import annotations + +import builtins + +import httpx +import pytest +from lfx.integrations import ( + INTEGRATION_ERROR_CODES, + AuthExpiredError, + ConnectionNotAuthorizedError, + IntegrationError, + ProviderUnavailableError, + RateLimitedError, + ScopeMissingError, + normalize_integration_error, + register_error_normalizer, +) + + +@pytest.fixture(autouse=True) +def isolated_normalizers(monkeypatch: pytest.MonkeyPatch) -> None: + from lfx.integrations import errors + + monkeypatch.setattr(errors, "_NORMALIZERS", {}) + + +def _http_error(status: int) -> httpx.HTTPStatusError: + request = httpx.Request("GET", "https://provider.example/private?token=secret") + response = httpx.Response(status, request=request) + return httpx.HTTPStatusError("provider rejected user@example.com", request=request, response=response) + + +def test_integration_error_codes_are_stable() -> None: + assert { + "connection-unresolved", + "connection-not-authorized", + "auth-expired", + "scope-missing", + "rate-limited", + "provider-unavailable", + "action-unsupported", + } == INTEGRATION_ERROR_CODES + + +@pytest.mark.parametrize( + ("status", "error_type"), + [ + (401, AuthExpiredError), + (403, ConnectionNotAuthorizedError), + (429, RateLimitedError), + (500, ProviderUnavailableError), + ], +) +def test_normalize_integration_error_maps_http_status(status: int, error_type: type[IntegrationError]) -> None: + assert isinstance(normalize_integration_error(_http_error(status), provider="google"), error_type) + + +def test_normalize_integration_error_unwraps_exception_groups() -> None: + group_type = getattr(builtins, "ExceptionGroup", None) + if group_type is None: + group_type = pytest.importorskip("exceptiongroup").ExceptionGroup + grouped = group_type("provider call", [RuntimeError("outer"), _http_error(401)]) + + assert isinstance(normalize_integration_error(grouped, provider="google"), AuthExpiredError) + + +def test_integration_error_sanitizes_urls_and_email() -> None: + error = IntegrationError( + "failed for user@example.com at https://example.com/path?token=secret", + details={"upstream": "https://example.com/private?token=secret"}, + ) + + rendered = str(error) + assert "user@example.com" not in rendered + assert "token=secret" not in rendered + assert "token=secret" not in error.details["upstream"] + + +def test_bundle_can_register_provider_error_normalizer() -> None: + register_error_normalizer("test-provider", lambda _exc: AuthExpiredError(provider="test-provider")) + + assert isinstance(normalize_integration_error(RuntimeError("unsafe"), provider="test-provider"), AuthExpiredError) + + +@pytest.mark.parametrize("chain_attribute", ["__cause__", "__context__"]) +@pytest.mark.parametrize(("status", "error_type"), [(401, AuthExpiredError), (429, RateLimitedError)]) +def test_normalize_chained_http_failures(chain_attribute: str, status: int, error_type: type[IntegrationError]) -> None: + inner = _http_error(status) + inner.response.headers["retry-after"] = "7" + outer = RuntimeError("wrapped") + setattr(outer, chain_attribute, inner) + normalized = normalize_integration_error(outer, provider="google") + + assert isinstance(normalized, error_type) + if status == 429: + assert normalized.retry_after == 7 + + +def test_normalize_chained_group_and_cycle() -> None: + group_type = getattr(builtins, "ExceptionGroup", None) + if group_type is None: + group_type = pytest.importorskip("exceptiongroup").ExceptionGroup + outer = RuntimeError("wrapped") + inner = _http_error(401) + outer.__cause__ = group_type("group", [RuntimeError("noise"), inner]) + inner.__context__ = outer + assert isinstance(normalize_integration_error(outer, provider="google"), AuthExpiredError) + cycle = RuntimeError("cycle") + cycle.__cause__ = cycle + assert isinstance(normalize_integration_error(cycle, provider="google"), ProviderUnavailableError) + + +def test_generic_provider_denial_does_not_claim_missing_scopes() -> None: + error = normalize_integration_error(_http_error(403), provider="google") + assert error.code == "connection-not-authorized" + assert "scope" not in error.hint.lower() + assert "provider" in str(error).lower() + + +def test_explicit_insufficient_scope_challenge_is_actionable() -> None: + error = _http_error(403) + error.response.headers["www-authenticate"] = 'Bearer error="insufficient_scope"' + assert isinstance(normalize_integration_error(error, provider="google"), ScopeMissingError) diff --git a/src/lfx/tests/unit/integrations/test_imports.py b/src/lfx/tests/unit/integrations/test_imports.py new file mode 100644 index 000000000000..5d1400a242dd --- /dev/null +++ b/src/lfx/tests/unit/integrations/test_imports.py @@ -0,0 +1,39 @@ +"""Connection schemas must not add provider client or telemetry imports.""" + +import subprocess +import sys + +import pytest + + +@pytest.mark.parametrize("module", ["lfx.integrations", "lfx.extension.manifest", "lfx.inputs.input_mixin"]) +def test_schema_imports_do_not_load_mcp_or_observability(module: str) -> None: + result = subprocess.run( # noqa: S603 - fixed interpreter and parametrized local module names + [ + sys.executable, + "-c", + "import importlib, sys; " + f"importlib.import_module({module!r}); " + "assert 'mcp' not in sys.modules; " + "assert 'lfx.base.mcp.util' not in sys.modules; " + + ( + "assert 'lfx.observability' not in sys.modules; assert 'langchain_core' not in sys.modules" + if module != "lfx.inputs.input_mixin" + else "" # Inputs already import observability on the release base. + ), + ], + capture_output=True, + text=True, + timeout=30, + check=False, + ) + assert result.returncode == 0, result.stderr + + +def test_public_exports_remain_available() -> None: + from lfx import integrations + + for name in integrations.__all__: + assert getattr(integrations, name) is not None + with pytest.raises(AttributeError): + _ = integrations.not_a_public_contract diff --git a/src/lfx/tests/unit/integrations/test_runtime_wiring.py b/src/lfx/tests/unit/integrations/test_runtime_wiring.py new file mode 100644 index 000000000000..b12b72885c08 --- /dev/null +++ b/src/lfx/tests/unit/integrations/test_runtime_wiring.py @@ -0,0 +1,292 @@ +from __future__ import annotations + +import asyncio +import copy +from types import SimpleNamespace +from typing import TYPE_CHECKING + +import pytest +from lfx.cli.validation import validate_connection_refs_for_env +from lfx.custom.custom_component.component import Component +from lfx.graph.graph.base import Graph +from lfx.inputs.inputs import ConnectionRefInput +from lfx.integrations import ( + ConditionalScopeRequirement, + ConnectionUnresolvedError, + ResolvedCredential, + ScopeCondition, + integration_action, +) +from lfx.io import BoolInput +from lfx.run._defaults import apply_run_defaults +from lfx.services.authorization.base import ExecutionPrincipal +from lfx.services.schema import ServiceType +from pydantic import SecretStr + +if TYPE_CHECKING: + from pathlib import Path + + +class ConnectionComponent(Component): + inputs = [ + ConnectionRefInput( + name="connection", + provider="google", + required_scopes=["drive.read"], + ) + ] + + +class CapturingResolver: + def __init__(self) -> None: + self.request = None + + async def resolve(self, request): + self.request = request + return ResolvedCredential(access_token=SecretStr("runtime-token"), provider="google", name="work") + + async def describe(self, _ref, _principal): + return None + + +def test_headless_principal_is_in_memory_and_propagated_to_graph_copies() -> None: + graph = Graph() + apply_run_defaults(graph, session_id="session-1", user_id="operator-1") + + graph_copy = copy.deepcopy(graph) + + assert graph.execution_principal.kind == "headless_operator" + assert graph_copy.execution_principal == graph.execution_principal + assert "execution_principal" not in graph.__getstate__() + + +@pytest.mark.asyncio +async def test_component_builds_lazy_lease_from_graph_principal(monkeypatch: pytest.MonkeyPatch) -> None: + resolver = CapturingResolver() + monkeypatch.setattr("lfx.services.deps.get_connection_resolver", lambda: resolver) + component = ConnectionComponent(connection="google/work") + graph = SimpleNamespace( + execution_principal=ExecutionPrincipal(kind="actor", user_id="user-1", interactive=True), + flow_id="flow-1", + run_id="run-1", + ) + component.set_vertex(SimpleNamespace(graph=graph)) + + lease = component.resolve_connection("connection") + + assert resolver.request is None + assert await lease.get_token() == "runtime-token" + assert resolver.request.principal.user_id == "user-1" + assert resolver.request.required_scopes == frozenset({"drive.read"}) + + +@pytest.mark.parametrize("no_env_fallback", [False, True]) +def test_headless_preflight_reports_missing_connection(monkeypatch: pytest.MonkeyPatch, no_env_fallback) -> None: + monkeypatch.delenv("LF_CONNECTION__TEST_2EPROVIDER__WORK", raising=False) + vertex = SimpleNamespace( + data={ + "node": { + "template": { + "connection": {"type": "connection_ref", "value": "test.provider/work"}, + } + } + }, + params={"connection": "test.provider/work"}, + ) + graph = SimpleNamespace(vertices=[vertex], context={"no_env_fallback": no_env_fallback}) + + errors = validate_connection_refs_for_env(graph) + + assert len(errors) == 1 + assert isinstance(errors[0], ConnectionUnresolvedError) + assert errors[0].env_key == "LF_CONNECTION__TEST_2EPROVIDER__WORK" + assert errors[0].reason == ("env-fallback-disabled" if no_env_fallback else "missing") + if no_env_fallback: + assert "Set LF_CONNECTION" not in str(errors[0]) + + +@pytest.mark.asyncio +async def test_integration_telemetry_excludes_connection_identifiers(monkeypatch: pytest.MonkeyPatch) -> None: + captured = [] + + class Telemetry: + async def log_integration_action(self, payload): + captured.append((payload, "integration_action")) + + manager = SimpleNamespace(services={ServiceType.TELEMETRY_SERVICE: Telemetry()}) + monkeypatch.setattr("lfx.services.manager.get_service_manager", lambda: manager) + component = SimpleNamespace( + graph=SimpleNamespace(execution_principal=ExecutionPrincipal(kind="headless_operator")), + log=lambda *_args, **_kwargs: None, + ) + + async with integration_action(component, provider="google", capability="drive.read", owner_kind="env"): + pass + + payload, event_name = captured[0] + rendered = payload.model_dump() + assert event_name == "integration_action" + assert set(rendered) == { + "client_type", + "provider", + "capability", + "ms", + "success", + "error_code", + "owner_kind", + "principal_kind", + } + assert "connection" not in rendered + + +@pytest.mark.parametrize("write", [False, True]) +async def test_real_graph_lease_before_run_enforces_active_conditional_scopes( + monkeypatch: pytest.MonkeyPatch, + write: bool, # noqa: FBT001 - parametrized conditional input +) -> None: + class ConditionalComponent(Component): + inputs = [ + BoolInput(name="write", value=False), + ConnectionRefInput( + name="connection", + provider="google", + required_scopes=["drive.read"], + conditional_scopes=[ + ConditionalScopeRequirement( + scope="drive.write", + role="optional", + condition=ScopeCondition(kind="input_truthy", input="write"), + ) + ], + ), + ] + + resolver = CapturingResolver() + monkeypatch.setattr("lfx.services.deps.get_connection_resolver", lambda: resolver) + graph = Graph() + apply_run_defaults(graph, session_id="session", user_id="operator") + component = ConditionalComponent(connection="google/work", write=write) + component.set_vertex(SimpleNamespace(graph=graph)) + + lease = component.resolve_connection("connection") + assert resolver.request is None + assert await lease.get_token() == "runtime-token" + assert resolver.request.run_id is None + assert resolver.request.required_scopes == frozenset({"drive.read", "drive.write"} if write else {"drive.read"}) + + +async def test_subgraph_inherits_execution_principal() -> None: + from lfx.components.input_output import TextOutputComponent + from lfx.services.authorization.base import ExecutionPrincipal + + component = TextOutputComponent(_id="output").set(input_value="hello") + graph = Graph(start=component, end=component) + graph.execution_principal = ExecutionPrincipal(kind="actor", user_id="actor", family="v1_run", interactive=True) + + async with graph.create_subgraph({"output"}) as subgraph: + assert subgraph.execution_principal == graph.execution_principal + assert "execution_principal" not in subgraph.__getstate__() + + +def test_malformed_preflight_never_echoes_field_value() -> None: + raw = "sensitive-token-value" + graph = SimpleNamespace( + vertices=[ + SimpleNamespace( + data={"node": {"template": {"connection": {"type": "connection_ref", "value": raw}}}}, params={} + ) + ], + context={}, + ) + errors = validate_connection_refs_for_env(graph) + assert len(errors) == 1 + assert errors[0].code == "connection-unresolved" + assert raw not in str(errors[0]) + assert raw not in repr(vars(errors[0])) + + +async def test_integration_action_does_not_wait_for_telemetry_transport(monkeypatch: pytest.MonkeyPatch) -> None: + from lfx.services.telemetry.service import TelemetryService + + service = TelemetryService(do_not_track=False) + entered = asyncio.Event() + release = asyncio.Event() + sent = [] + + async def send(payload, path): + sent.append((payload, path)) + entered.set() + await release.wait() + + monkeypatch.setattr(service, "send_telemetry_data", send) + manager = SimpleNamespace(services={ServiceType.TELEMETRY_SERVICE: service}) + monkeypatch.setattr("lfx.services.manager.get_service_manager", lambda: manager) + component = SimpleNamespace(graph=None, log=lambda *_args, **_kwargs: None) + + async def action(): + async with integration_action(component, provider="google", capability="drive.read", owner_kind="env"): + pass + + service.start() + try: + await asyncio.wait_for(action(), timeout=1) + await asyncio.wait_for(entered.wait(), timeout=1) + assert sent[0][1] == "integration_action" + assert not release.is_set() + finally: + release.set() + await service.stop() + + +async def test_run_flow_uses_configured_resolver_without_environment( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + from lfx.run.base import run_flow + from lfx.services.connection.base import BaseConnectionResolverService, ConnectionAccessPolicy + from lfx.services.manager import get_service_manager + + class HostResolver(BaseConnectionResolverService): + def __init__(self): + super().__init__() + self.set_ready() + + async def _get_access_policy(self, _request): + return ConnectionAccessPolicy(owner_kind="env", allow_non_interactive=True) + + async def _resolve(self, request, _policy): + return ResolvedCredential(access_token=SecretStr("host-only"), provider=request.ref.provider) + + manager = get_service_manager() + monkeypatch.setitem(manager.service_classes, ServiceType.CONNECTION_RESOLVER_SERVICE, HostResolver) + monkeypatch.delitem(manager.services, ServiceType.CONNECTION_RESOLVER_SERVICE, raising=False) + monkeypatch.delenv("LF_CONNECTION__GOOGLE__WORK", raising=False) + script = tmp_path / "connection_flow.py" + script.write_text( + """from lfx.custom import Component +from lfx.components.input_output import ChatInput, ChatOutput +from lfx.graph import Graph +from lfx.io import ConnectionRefInput, MessageTextInput, Output +from lfx.schema.message import Message + +class ConnectionProbe(Component): + inputs = [MessageTextInput(name="input_value"), ConnectionRefInput(name="connection", provider="google")] + outputs = [Output(name="result", method="resolve", display_name="Result")] + + async def resolve(self) -> Message: + token = await self.resolve_connection("connection").get_token() + return Message(text="resolved" if token == "host-only" else "incorrect credential") + +chat = ChatInput(_id="chat").set(input_value="hello") +component = ConnectionProbe(_id="probe", connection="google/work").set(input_value=chat.message_response) +output = ChatOutput(_id="output").set(input_value=component.resolve) +graph = Graph(start=chat, end=output) +""", + encoding="utf-8", + ) + try: + result = await run_flow(script_path=script, check_variables=True) + assert result["success"] is True + assert "resolved" in str(result) + assert "host-only" not in str(result) + finally: + manager.services.pop(ServiceType.CONNECTION_RESOLVER_SERVICE, None) diff --git a/src/lfx/tests/unit/services/connection/__init__.py b/src/lfx/tests/unit/services/connection/__init__.py new file mode 100644 index 000000000000..275e6aa0d2ae --- /dev/null +++ b/src/lfx/tests/unit/services/connection/__init__.py @@ -0,0 +1 @@ +"""Connection resolver service tests.""" diff --git a/src/lfx/tests/unit/services/connection/test_credential_lease.py b/src/lfx/tests/unit/services/connection/test_credential_lease.py new file mode 100644 index 000000000000..28c143b302ba --- /dev/null +++ b/src/lfx/tests/unit/services/connection/test_credential_lease.py @@ -0,0 +1,96 @@ +from __future__ import annotations + +import asyncio +from datetime import datetime, timedelta, timezone + +import pytest +from lfx.integrations import ( + AuthExpiredError, + ConnectionRef, + ConnectionResolutionRequest, + CredentialLease, + ResolvedCredential, +) +from lfx.services.authorization.base import ExecutionPrincipal +from pydantic import SecretStr + + +class Resolver: + def __init__(self, credentials: list[ResolvedCredential]) -> None: + self.credentials = credentials + self.calls = 0 + + async def resolve(self, _request: ConnectionResolutionRequest) -> ResolvedCredential: + await asyncio.sleep(0) + credential = self.credentials[min(self.calls, len(self.credentials) - 1)] + self.calls += 1 + return credential + + async def describe(self, _ref, _principal): + return None + + +class FailingResolver(Resolver): + async def resolve(self, request: ConnectionResolutionRequest) -> ResolvedCredential: + self.calls += 1 + if self.calls > 1: + raise AuthExpiredError(provider=request.ref.provider) + return self.credentials[0] + + +def _credential(token: str, expires_at: datetime | None = None) -> ResolvedCredential: + return ResolvedCredential(access_token=SecretStr(token), expires_at=expires_at, provider="google", name="work") + + +def _request() -> ConnectionResolutionRequest: + return ConnectionResolutionRequest( + ref=ConnectionRef.parse("google/work"), + principal=ExecutionPrincipal(kind="headless_operator"), + ) + + +@pytest.mark.asyncio +async def test_initial_resolution_is_single_flight() -> None: + resolver = Resolver([_credential("token")]) + lease = CredentialLease(resolver, _request()) + + assert await asyncio.gather(*(lease.get_token() for _ in range(10))) == ["token"] * 10 + assert resolver.calls == 1 + + +@pytest.mark.asyncio +async def test_expiring_credential_is_refreshed() -> None: + now = datetime.now(timezone.utc) + resolver = Resolver([_credential("old", now + timedelta(seconds=30)), _credential("new", now + timedelta(hours=1))]) + lease = CredentialLease(resolver, _request(), now=lambda: now) + + assert await lease.get_token() == "old" + assert await lease.get_token() == "new" + assert resolver.calls == 2 + + +@pytest.mark.asyncio +async def test_no_expiry_credential_refreshes_reactively_once() -> None: + resolver = Resolver([_credential("old"), _credential("new")]) + lease = CredentialLease(resolver, _request()) + error = AuthExpiredError(provider="google") + + assert await lease.get_token() == "old" + assert await lease.get_token_after_auth_error(error) == "new" + with pytest.raises(AuthExpiredError): + await lease.get_token_after_auth_error(error) + assert resolver.calls == 2 + + +@pytest.mark.asyncio +async def test_failed_reactive_refresh_is_not_retried() -> None: + resolver = FailingResolver([_credential("old")]) + lease = CredentialLease(resolver, _request()) + error = AuthExpiredError(provider="google") + + assert await lease.get_token() == "old" + with pytest.raises(AuthExpiredError): + await lease.get_token_after_auth_error(error) + with pytest.raises(AuthExpiredError): + await lease.get_token_after_auth_error(error) + assert resolver.calls == 2 diff --git a/src/lfx/tests/unit/services/connection/test_discovery.py b/src/lfx/tests/unit/services/connection/test_discovery.py new file mode 100644 index 000000000000..9e499204f0b4 --- /dev/null +++ b/src/lfx/tests/unit/services/connection/test_discovery.py @@ -0,0 +1,315 @@ +from __future__ import annotations + +from types import SimpleNamespace +from typing import TYPE_CHECKING, Literal + +import pytest +from lfx.integrations import ( + ConnectionNotAuthorizedError, + ConnectionRef, + ConnectionResolutionRequest, + ResolvedCredential, + ScopeMissingError, +) +from lfx.services.authorization.base import ExecutionPrincipal +from lfx.services.connection.base import BaseConnectionResolverService, ConnectionAccessPolicy +from lfx.services.connection.env_resolver import EnvConnectionResolver +from lfx.services.deps import get_connection_resolver +from lfx.services.factory import ServiceFactory +from lfx.services.manager import ServiceManager +from lfx.services.schema import ServiceType +from pydantic import SecretStr + +if TYPE_CHECKING: + from pathlib import Path + + +def _request(principal: ExecutionPrincipal) -> ConnectionResolutionRequest: + return ConnectionResolutionRequest(ref=ConnectionRef.parse("google/work"), principal=principal) + + +class PolicyResolver(BaseConnectionResolverService): + """A host that deliberately omits the portable authorization helper.""" + + def __init__(self, policy: ConnectionAccessPolicy): + super().__init__() + self.policy = policy + self.policy_reads = 0 + self.credential_reads = 0 + self.credential = ResolvedCredential(access_token=SecretStr("host-token"), owner_kind=policy.owner_kind) + self.set_ready() + + def authorize_principal(self, *_args, **_kwargs): + # Even an override cannot weaken the base resolve entry point. + return None + + async def _get_access_policy(self, _request): + self.policy_reads += 1 + return self.policy + + async def _resolve(self, _request, policy): + assert policy is self.policy + self.credential_reads += 1 + return self.credential + + +async def _assert_resolution(resolver: PolicyResolver, principal: ExecutionPrincipal, *, allowed: bool) -> None: + if allowed: + assert (await resolver.resolve(_request(principal))).access_token.get_secret_value() == "host-token" + else: + with pytest.raises(ConnectionNotAuthorizedError): + await resolver.resolve(_request(principal)) + assert resolver.credential_reads == int(allowed) + if principal.kind in {"unknown", "anonymous_public"}: + assert resolver.policy_reads == 0 + + +@pytest.mark.parametrize( + ("principal", "owner_kind", "owner_id", "allow_non_interactive", "allowed"), + [ + (ExecutionPrincipal(kind="headless_operator"), "env", None, True, True), + (ExecutionPrincipal(kind="actor", user_id="user-1", interactive=True), "env", None, True, False), + (ExecutionPrincipal(kind="unknown"), "instance", None, True, False), + (ExecutionPrincipal(kind="anonymous_public"), "instance", None, True, False), + (ExecutionPrincipal(kind="actor", user_id="user-1", interactive=True), "instance", None, True, True), + (ExecutionPrincipal(kind="actor", user_id="user-1", interactive=True), "user", "user-1", False, True), + (ExecutionPrincipal(kind="actor", user_id="user-1", interactive=True), "user", None, False, False), + (ExecutionPrincipal(kind="actor", user_id="user-1", interactive=True), "user", "user-2", False, False), + (ExecutionPrincipal(kind="flow_owner", user_id="user-1"), "user", "user-1", False, False), + (ExecutionPrincipal(kind="flow_owner", user_id="user-1"), "user", "user-1", True, True), + ], +) +async def test_portable_principal_authorization_floor( + principal: ExecutionPrincipal, + owner_kind: Literal["user", "instance", "env"], + owner_id: str | None, + allow_non_interactive: bool, # noqa: FBT001 - parametrized contract dimension + allowed: bool, # noqa: FBT001 - expected authorization result +) -> None: + resolver = PolicyResolver( + ConnectionAccessPolicy( + connection_owner_id=owner_id, owner_kind=owner_kind, allow_non_interactive=allow_non_interactive + ) + ) + await _assert_resolution(resolver, principal, allowed=allowed) + + +def test_configured_resolver_with_wrong_base_fails_closed() -> None: + manager = ServiceManager() + + with pytest.raises(RuntimeError, match="must subclass BaseConnectionResolverService"): + manager._register_service_from_path("connection_resolver_service", "builtins:str") + + +def test_absent_resolver_uses_headless_fallback(monkeypatch: pytest.MonkeyPatch) -> None: + manager = ServiceManager() + manager._plugins_discovered = True + monkeypatch.setattr("lfx.services.manager.get_service_manager", lambda: manager) + + resolver = get_connection_resolver() + + assert isinstance(resolver, EnvConnectionResolver) + assert ServiceType.CONNECTION_RESOLVER_SERVICE not in manager.services + assert get_connection_resolver() is resolver + + +class HostResolver(EnvConnectionResolver): + """Distinct host implementation for discovery and registration tests.""" + + +@pytest.mark.parametrize("registration", ["class", "factory"]) +def test_late_resolver_registration_replaces_environment_fallback(monkeypatch: pytest.MonkeyPatch, registration: str): + manager = ServiceManager() + manager._plugins_discovered = True + monkeypatch.setattr("lfx.services.manager.get_service_manager", lambda: manager) + fallback = get_connection_resolver() + + if registration == "class": + manager.register_service_class(ServiceType.CONNECTION_RESOLVER_SERVICE, HostResolver) + else: + + class HostFactory(ServiceFactory): + def __init__(self): + super().__init__() + self.service_class = HostResolver + + def create(self): + return HostResolver() + + manager.register_factory(HostFactory()) + + assert isinstance(get_connection_resolver(), HostResolver) + assert get_connection_resolver() is not fallback + + +@pytest.mark.parametrize("failure", ["wrong_base", "import", "attribute", "value", "resolve_override"]) +def test_entry_point_failures_never_select_environment_fallback( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path, failure: str +) -> None: + def load(): + if failure == "wrong_base": + return str + if failure == "resolve_override": + + class UnsafeResolver(EnvConnectionResolver): + async def resolve(self, _request): + return ResolvedCredential(access_token=SecretStr("unguarded-token"), owner_kind="user") + + return UnsafeResolver + msg = "broken plugin" + raise {"import": ImportError, "attribute": AttributeError, "value": ValueError}[failure](msg) + + ep = SimpleNamespace(name=ServiceType.CONNECTION_RESOLVER_SERVICE.value, load=load) + monkeypatch.setattr("importlib.metadata.entry_points", lambda **_kwargs: [ep]) + manager = ServiceManager() + monkeypatch.setattr("lfx.services.manager.get_service_manager", lambda: manager) + with pytest.raises(RuntimeError, match="refusing"): + manager.discover_plugins(tmp_path) + assert manager._plugins_discovered is False + with pytest.raises(RuntimeError, match="refusing"): + get_connection_resolver() + assert ServiceType.CONNECTION_RESOLVER_SERVICE not in manager.services + + +def test_missing_configured_resolver_path_fails_closed(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + (tmp_path / "lfx.toml").write_text('[services]\nconnection_resolver_service = "missing_resolver:Service"\n') + monkeypatch.setattr("importlib.metadata.entry_points", lambda **_kwargs: []) + with pytest.raises(RuntimeError, match="refusing"): + ServiceManager().discover_plugins(tmp_path) + + +@pytest.mark.parametrize("failure", ["constructor", "not_ready", "missing_factory"]) +def test_broken_registered_resolver_never_falls_back(monkeypatch: pytest.MonkeyPatch, failure: str) -> None: + from lfx.services.manager import NoFactoryRegisteredError + + class BrokenResolver(EnvConnectionResolver): + def __init__(self): + if failure == "constructor": + msg = "constructor failed" + raise RuntimeError(msg) + if failure == "missing_factory": + msg = "dependency is absent" + raise NoFactoryRegisteredError(msg) + super().__init__() + self._ready = False + + manager = ServiceManager() + manager._plugins_discovered = True + monkeypatch.setattr("lfx.services.manager.get_service_manager", lambda: manager) + manager.register_service_class(ServiceType.CONNECTION_RESOLVER_SERVICE, BrokenResolver) + with pytest.raises((RuntimeError, NoFactoryRegisteredError)): + get_connection_resolver() + + +@pytest.mark.parametrize( + ("principal", "owner_kind", "owner_id", "allow_non_interactive", "allowed"), + [ + (ExecutionPrincipal(kind="actor", user_id="actor", interactive=True), "user", "owner", False, True), + (ExecutionPrincipal(kind="actor", user_id="actor", interactive=True), "user", None, True, False), + (ExecutionPrincipal(kind="actor", interactive=True), "user", "owner", True, False), + (ExecutionPrincipal(kind="actor", user_id="actor"), "user", "owner", False, False), + (ExecutionPrincipal(kind="flow_owner", user_id="actor"), "user", "owner", True, False), + (ExecutionPrincipal(kind="anonymous_public", user_id="actor"), "user", "owner", True, False), + (ExecutionPrincipal(kind="unknown", user_id="actor"), "user", "owner", True, False), + (ExecutionPrincipal(kind="actor", user_id="actor", interactive=True), "env", None, True, False), + ], +) +async def test_verified_share_only_satisfies_actor_ownership_mismatch( + principal: ExecutionPrincipal, + owner_kind: Literal["user", "instance", "env"], + owner_id: str | None, + allow_non_interactive: bool, # noqa: FBT001 - authorization contract dimension + allowed: bool, # noqa: FBT001 - expected result +) -> None: + resolver = PolicyResolver( + ConnectionAccessPolicy( + connection_owner_id=owner_id, + owner_kind=owner_kind, + allow_non_interactive=allow_non_interactive, + explicit_share_authorized=True, + ) + ) + await _assert_resolution(resolver, principal, allowed=allowed) + + +@pytest.mark.parametrize("base", [BaseConnectionResolverService, EnvConnectionResolver]) +def test_resolve_override_is_rejected_at_class_definition(base): + with pytest.raises(TypeError, match="resolve cannot be overridden"): + + class UnsafeResolver(base): + async def resolve(self, _request): + return ResolvedCredential(access_token=SecretStr("unguarded-token"), owner_kind="user") + + +def test_mixin_cannot_replace_guarded_entry_point(): + class UnsafeMixin: + async def resolve(self, _request): + return ResolvedCredential(access_token=SecretStr("unguarded-token"), owner_kind="user") + + with pytest.raises(TypeError, match="resolve cannot be overridden"): + + class UnsafeResolver(UnsafeMixin, EnvConnectionResolver): + pass + + +def test_host_cannot_omit_policy_hook(): + class IncompleteResolver(BaseConnectionResolverService): + async def _resolve(self, _request, _policy): + return ResolvedCredential(access_token=SecretStr("unguarded-token")) + + with pytest.raises(TypeError, match="_get_access_policy"): + IncompleteResolver() + + +async def test_invalid_host_policy_never_reads_credentials(): + resolver = PolicyResolver(ConnectionAccessPolicy(owner_kind="user")) + resolver.policy = None + await _assert_resolution( + resolver, ExecutionPrincipal(kind="actor", user_id="owner", interactive=True), allowed=False + ) + + +@pytest.mark.parametrize( + ("scopes_verified", "granted", "allowed"), + [ + (False, frozenset({"drive.readonly"}), False), + (True, frozenset(), False), + (True, frozenset({"drive.readonly"}), True), + ], +) +async def test_host_resolver_cannot_skip_scope_enforcement(scopes_verified, granted, allowed): + resolver = PolicyResolver(ConnectionAccessPolicy(owner_kind="user", connection_owner_id="owner")) + resolver.credential = ResolvedCredential( + access_token=SecretStr("host-token"), + owner_kind="user", + scopes_verified=scopes_verified, + granted_scopes=granted, + ) + request = ConnectionResolutionRequest( + ref=ConnectionRef.parse("google/work"), + principal=ExecutionPrincipal(kind="actor", user_id="owner", interactive=True), + required_scopes=frozenset({"https://www.googleapis.com/auth/drive.readonly"}), + ) + if allowed: + assert await resolver.resolve(request) is resolver.credential + else: + with pytest.raises(ScopeMissingError) as caught: + await resolver.resolve(request) + assert caught.value.details["scopes_verified"] is scopes_verified + assert resolver.credential_reads == 1 + + +async def test_manager_teardown_disposes_separate_fallback(monkeypatch: pytest.MonkeyPatch) -> None: + manager = ServiceManager() + manager._plugins_discovered = True + monkeypatch.setattr("lfx.services.manager.get_service_manager", lambda: manager) + fallback = get_connection_resolver() + disposed = [] + + async def teardown(): + disposed.append(True) + + monkeypatch.setattr(fallback, "teardown", teardown) + await manager.teardown() + assert disposed == [True] + assert manager.connection_resolver_fallback is None diff --git a/src/lfx/tests/unit/services/connection/test_env_resolver.py b/src/lfx/tests/unit/services/connection/test_env_resolver.py new file mode 100644 index 000000000000..52be62dd5eab --- /dev/null +++ b/src/lfx/tests/unit/services/connection/test_env_resolver.py @@ -0,0 +1,226 @@ +from __future__ import annotations + +import json +import traceback +from datetime import datetime, timedelta, timezone + +import pytest +from lfx.integrations import ( + AuthExpiredError, + ConnectionNotAuthorizedError, + ConnectionRef, + ConnectionResolutionRequest, + ConnectionUnresolvedError, + ScopeMissingError, +) +from lfx.services.authorization.base import ExecutionPrincipal +from lfx.services.connection.env_resolver import EnvConnectionResolver +from lfx.services.variable.request_scope import ( + activate_no_env_fallback, + activate_request_variables, + reset_no_env_fallback, + reset_request_variables, +) +from lfx.services.variable.service import VariableService + + +def _request(*, scopes: frozenset[str] = frozenset()) -> ConnectionResolutionRequest: + return ConnectionResolutionRequest( + ref=ConnectionRef.parse("google/work"), + principal=ExecutionPrincipal(kind="headless_operator"), + required_scopes=scopes, + ) + + +@pytest.fixture +def variable_service(monkeypatch: pytest.MonkeyPatch) -> VariableService: + service = VariableService() + monkeypatch.setattr("lfx.services.deps.get_variable_service", lambda: service) + return service + + +@pytest.mark.asyncio +async def test_request_scope_beats_environment( + monkeypatch: pytest.MonkeyPatch, + variable_service: VariableService, +) -> None: + _ = variable_service + env_key = _request().ref.env_key() + monkeypatch.setenv(env_key, "ambient-token") + token = activate_request_variables({env_key: "request-token"}) + try: + credential = await EnvConnectionResolver().resolve(_request()) + finally: + reset_request_variables(token) + + assert credential.access_token.get_secret_value() == "request-token" + + +@pytest.mark.asyncio +async def test_no_env_fallback_blocks_ambient_connection( + monkeypatch: pytest.MonkeyPatch, + variable_service: VariableService, +) -> None: + _ = variable_service + env_key = _request().ref.env_key() + monkeypatch.setenv(env_key, "ambient-token") + token = activate_no_env_fallback(disabled=True) + try: + with pytest.raises(ConnectionUnresolvedError) as caught: + await EnvConnectionResolver().resolve(_request()) + finally: + reset_no_env_fallback(token) + assert caught.value.details["reason"] == "env-fallback-disabled" + assert "request" in caught.value.hint + assert "Set LF_CONNECTION" not in str(caught.value) + + +@pytest.mark.asyncio +async def test_json_wire_format_and_scopes(variable_service: VariableService) -> None: + request = _request(scopes=frozenset({"drive.read"})) + variable_service.set_variable( + request.ref.env_key(), + json.dumps( + { + "access_token": "short-lived", + "token_type": "Bearer", + "expires_at": (datetime.now(timezone.utc) + timedelta(hours=1)).isoformat(), + "scopes": ["drive.read"], + "account": {"id": "acct-1", "display": "Work"}, + } + ), + ) + + credential = await EnvConnectionResolver().resolve(request) + + assert credential.scopes_verified is True + assert credential.account is not None + assert credential.account.id == "acct-1" + + +@pytest.mark.asyncio +async def test_missing_and_scope_failures_are_typed(variable_service: VariableService) -> None: + resolver = EnvConnectionResolver() + with pytest.raises(ConnectionUnresolvedError): + await resolver.resolve(_request()) + + variable_service.set_variable(_request().ref.env_key(), json.dumps({"access_token": "token", "scopes": []})) + with pytest.raises(ScopeMissingError): + await resolver.resolve(_request(scopes=frozenset({"drive.read"}))) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("field", ["refresh_token", "client_secret", "password"]) +async def test_long_lived_secrets_are_rejected(variable_service: VariableService, field: str) -> None: + variable_service.set_variable( + _request().ref.env_key(), + json.dumps({"access_token": "token", field: "must-not-enter-runtime"}), + ) + + with pytest.raises(ConnectionUnresolvedError) as caught: + await EnvConnectionResolver().resolve(_request()) + assert caught.value.details["reason"] == "long-lived-secret" + assert "refresh_token" in caught.value.hint + assert caught.value.__context__ is None + assert caught.value.__cause__ is None + assert "must-not-enter-runtime" not in repr(vars(caught.value)) + assert "must-not-enter-runtime" not in "".join(traceback.format_exception(caught.value)) + + +@pytest.mark.parametrize("raw", ["token", '{"access_token":"token"}']) +async def test_required_scopes_reject_unverified_credentials(variable_service: VariableService, raw: str) -> None: + request = _request(scopes=frozenset({"drive.read"})) + variable_service.set_variable(request.ref.env_key(), raw) + with pytest.raises(ScopeMissingError) as caught: + await EnvConnectionResolver().resolve(request) + assert caught.value.details["scopes_verified"] is False + assert "scopes" in caught.value.hint + + +@pytest.mark.parametrize("raw", ["token", '{"access_token":"token"}']) +async def test_unverified_credentials_work_without_required_scopes(variable_service: VariableService, raw: str) -> None: + variable_service.set_variable(_request().ref.env_key(), raw) + credential = await EnvConnectionResolver().resolve(_request()) + assert credential.access_token.get_secret_value() == "token" + assert credential.scopes_verified is False + + +@pytest.mark.asyncio +async def test_expired_credential_is_typed(variable_service: VariableService) -> None: + variable_service.set_variable( + _request().ref.env_key(), + json.dumps( + { + "access_token": "expired", + "expires_at": (datetime.now(timezone.utc) - timedelta(minutes=1)).isoformat(), + } + ), + ) + + with pytest.raises(AuthExpiredError): + await EnvConnectionResolver().resolve(_request()) + + +@pytest.mark.asyncio +async def test_non_headless_principal_cannot_use_environment_connection(variable_service: VariableService) -> None: + request = ConnectionResolutionRequest( + ref=ConnectionRef.parse("google/work"), + principal=ExecutionPrincipal(kind="actor", user_id="user-1", interactive=True), + ) + variable_service.set_variable(request.ref.env_key(), "token") + + with pytest.raises(ConnectionNotAuthorizedError): + await EnvConnectionResolver().resolve(request) + + +@pytest.mark.parametrize( + ("invalid_fields", "reason"), + [ + ({"account": {"id": "ok", "id_token": "sensitive-wire-value"}}, "invalid-account"), + ({"account": "sensitive-wire-value"}, "invalid-account"), + ({"expires_at": "sensitive-wire-value"}, "invalid-expiry"), + ({"expires_at": 1e300}, "invalid-expiry"), + ({"scopes": {"invalid": "sensitive-wire-value"}}, "invalid-scopes"), + ({"token_type": ["sensitive-wire-value"]}, "invalid-token-type"), + ({"sensitive-wire-value": "unknown field name"}, "unsupported-fields"), + ({"access_token": ""}, "invalid-access-token"), + ], +) +async def test_malformed_credentials_are_typed_without_raw_exception_context( + variable_service: VariableService, invalid_fields: dict, reason: str +) -> None: + payload = {"access_token": "sensitive-wire-value", **invalid_fields} + variable_service.set_variable(_request().ref.env_key(), json.dumps(payload)) + + with pytest.raises(ConnectionUnresolvedError) as caught: + await EnvConnectionResolver().resolve(_request()) + + error = caught.value + assert error.reason == reason + assert error.__cause__ is None + assert error.__context__ is None + assert "sensitive-wire-value" not in "".join(traceback.format_exception(error)) + assert "sensitive-wire-value" not in repr(vars(error)) + assert error.env_key == _request().ref.env_key() + + +async def test_malformed_json_is_typed(variable_service: VariableService) -> None: + variable_service.set_variable(_request().ref.env_key(), '{"access_token":"sensitive-wire-value"') + with pytest.raises(ConnectionUnresolvedError) as caught: + await EnvConnectionResolver().resolve(_request()) + assert caught.value.__context__ is None + assert caught.value.reason == "malformed-json" + + +@pytest.mark.parametrize("provider", ["google", "google_workspace"]) +async def test_resolver_applies_shared_scope_normalization(variable_service: VariableService, provider: str) -> None: + request = ConnectionResolutionRequest( + ref=ConnectionRef.parse(f"{provider}/work"), + principal=ExecutionPrincipal(kind="headless_operator"), + required_scopes=frozenset({"https://www.googleapis.com/auth/drive.readonly"}), + ) + variable_service.set_variable( + request.ref.env_key(), json.dumps({"access_token": "token", "scopes": ["drive.readonly"]}) + ) + + assert (await EnvConnectionResolver().resolve(request)).scopes_verified diff --git a/src/lfx/tests/unit/services/connection/test_headless_execution.py b/src/lfx/tests/unit/services/connection/test_headless_execution.py new file mode 100644 index 000000000000..3c33d71ff097 --- /dev/null +++ b/src/lfx/tests/unit/services/connection/test_headless_execution.py @@ -0,0 +1,95 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING + +import pytest +from lfx.integrations import ScopeMissingError, normalize_integration_error +from lfx.run.base import RunError, run_flow +from lfx.services.variable.request_scope import ( + activate_no_env_fallback, + activate_request_variables, + get_active_request_variables, + is_env_fallback_disabled, + reset_no_env_fallback, + reset_request_variables, +) + +if TYPE_CHECKING: + from pathlib import Path + + +@pytest.fixture +def outer_scope(source, raw): + if source != "inherited_request": + yield + return + scope_token = activate_request_variables({"LF_CONNECTION__GOOGLE__WORK": raw}) + no_env_token = activate_no_env_fallback(disabled=True) + try: + yield + finally: + reset_no_env_fallback(no_env_token) + reset_request_variables(scope_token) + + +@pytest.mark.usefixtures("outer_scope") +@pytest.mark.parametrize("source", ["environment", "request", "inherited_request"]) +@pytest.mark.parametrize( + ("raw", "allowed"), + [ + ('{"access_token":"probe-token","scopes":["drive.read"]}', True), + ('{"access_token":"probe-token","scopes":["drive.write"]}', False), + ("probe-token", False), + ('{"access_token":"probe-token"}', False), + ], +) +async def test_headless_execution_enforces_declared_scopes(monkeypatch, tmp_path: Path, source, raw, allowed): + script = tmp_path / "scoped_connection.py" + script.write_text( + """from lfx.components.input_output import ChatInput, ChatOutput +from lfx.custom import Component +from lfx.graph import Graph +from lfx.io import ConnectionRefInput, MessageTextInput, Output +from lfx.schema.message import Message + +class ScopedConnectionProbe(Component): + inputs = [ + MessageTextInput(name="input_value"), + ConnectionRefInput(name="connection", provider="google", required_scopes=["drive.read"]), + ] + outputs = [Output(name="result", display_name="Result", method="check_connection")] + + async def check_connection(self) -> Message: + await self.resolve_connection("connection").get_token() + return Message(text="credential accepted") + +chat = ChatInput(_id="chat").set(input_value="hello") +probe = ScopedConnectionProbe(_id="probe", connection="google/work").set(input_value=chat.message_response) +output = ChatOutput(_id="output").set(input_value=probe.check_connection) +graph = Graph(start=chat, end=output) +""" + + ('graph.context["no_env_fallback"] = True\n' if source == "request" else ""), + encoding="utf-8", + ) + env_key = "LF_CONNECTION__GOOGLE__WORK" + # The request must win even when ambient credentials have sufficient scopes. + monkeypatch.setenv( + env_key, raw if source == "environment" else '{"access_token":"ambient","scopes":["drive.read"]}' + ) + if source == "inherited_request": + monkeypatch.delenv(env_key) + variables = {env_key: raw} if source == "request" else None + previous_scope = get_active_request_variables() + previous_no_env = is_env_fallback_disabled() + if allowed: + result = await run_flow(script_path=script, check_variables=True, global_variables=variables) + assert result["success"] is True + assert "credential accepted" in str(result) + assert "probe-token" not in str(result) + else: + with pytest.raises(RunError) as caught: + await run_flow(script_path=script, check_variables=True, global_variables=variables) + assert isinstance(normalize_integration_error(caught.value, provider="google"), ScopeMissingError) + assert "probe-token" not in str(caught.value) + assert get_active_request_variables() is previous_scope + assert is_env_fallback_disabled() is previous_no_env