fix(declarative): allow bundled custom components without AIRBYTE_ENABLE_UNSAFE_CODE

Only gate Custom* components when the manifest itself is supplied through the
config, which is the untrusted case. Manifests bundled in a published connector
image are trusted, so their custom components must keep working in environments
that do not set AIRBYTE_ENABLE_UNSAFE_CODE, such as Airbyte Cloud.

Co-Authored-By: syed.khadeer@airbyte.io <cloud-support@airbyte.io>

Author: devin-ai-integration[bot]

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -478,6 +478,7 @@
     ZipfileDecoder as ZipfileDecoderModel,
 )
 from airbyte_cdk.sources.declarative.parsers.custom_code_compiler import (
+    INJECTED_MANIFEST,
     AirbyteCustomCodeNotPermittedError,
     custom_code_execution_permitted,
 )
@@ -1814,11 +1815,11 @@ def create_custom_component(self, model: Any, config: Config, **kwargs: Any) ->
         :return: The declarative component built from the Pydantic model to be used at runtime
         """
         # Instantiating a custom component means importing and executing arbitrary code referenced
-        # by `class_name`. This is only permitted when custom code execution is explicitly enabled,
-        # mirroring the gate applied to injected `components.py` code. Without this check, a manifest
-        # could point `class_name` at any importable callable and have it invoked, bypassing the
-        # `AIRBYTE_ENABLE_UNSAFE_CODE` protection.
-        if not custom_code_execution_permitted():
+        # by `class_name`. When the manifest itself comes from the config, it is untrusted input and
+        # could point `class_name` at any importable callable, so it honors the same
+        # `AIRBYTE_ENABLE_UNSAFE_CODE` gate as injected `components.py` code. Manifests bundled in a
+        # published connector image are trusted and may always use their bundled custom components.
+        if config.get(INJECTED_MANIFEST) and not custom_code_execution_permitted():
             raise AirbyteCustomCodeNotPermittedError
 
         custom_component_class = self._get_class_from_fully_qualified_class_name(model.class_name)

unit_tests/legacy/sources/declarative/conftest.py

@@ -109,6 +109,7 @@
 )
 from airbyte_cdk.sources.declarative.parsers.custom_code_compiler import (
     ENV_VAR_ALLOW_CUSTOM_CODE,
+    INJECTED_MANIFEST,
     AirbyteCustomCodeNotPermittedError,
 )
 from airbyte_cdk.sources.declarative.parsers.manifest_component_transformer import (
@@ -2527,14 +2528,14 @@ def test_create_custom_components(manifest, field_name, expected_value, expected
     ],
 )
 def test_create_custom_component_requires_custom_code_enabled(class_name, monkeypatch):
-    """A `Custom*` component must not be instantiated unless custom code execution is
-    explicitly enabled via `AIRBYTE_ENABLE_UNSAFE_CODE`.
-
-    Resolving and instantiating a component's `class_name` executes arbitrary
-    importable code, so it must honor the same gate as injected `components.py` code.
-    The gate must fire regardless of whether `class_name` points at a bundled custom
-    component or at an arbitrary importable callable, and it must fire before the
-    referenced module is imported.
+    """A `Custom*` component declared by a config-injected manifest must not be instantiated
+    unless custom code execution is explicitly enabled via `AIRBYTE_ENABLE_UNSAFE_CODE`.
+
+    A manifest provided through the config is untrusted input, and resolving its
+    `class_name` executes arbitrary importable code, so it must honor the same gate as
+    injected `components.py` code. The gate must fire regardless of whether `class_name`
+    points at a bundled custom component or at an arbitrary importable callable, and it
+    must fire before the referenced module is imported.
     """
     monkeypatch.delenv(ENV_VAR_ALLOW_CUSTOM_CODE, raising=False)
 
@@ -2551,7 +2552,30 @@ def _fail_if_resolved(*args, **kwargs):
     }
 
     with pytest.raises(AirbyteCustomCodeNotPermittedError):
-        factory.create_component(CustomErrorHandlerModel, manifest, input_config)
+        factory.create_component(
+            CustomErrorHandlerModel,
+            manifest,
+            {**input_config, INJECTED_MANIFEST: {"type": "DeclarativeSource"}},
+        )
+
+
+def test_create_custom_component_permitted_for_bundled_manifest(monkeypatch):
+    """A manifest bundled in a connector image may use its bundled custom components.
+
+    Published manifest-only connectors ship their own `manifest.yaml` and `components.py`
+    inside a trusted image, so their `Custom*` components must keep working in environments
+    that do not set `AIRBYTE_ENABLE_UNSAFE_CODE`, such as Airbyte Cloud.
+    """
+    monkeypatch.delenv(ENV_VAR_ALLOW_CUSTOM_CODE, raising=False)
+
+    manifest = {
+        "type": "CustomErrorHandler",
+        "class_name": "unit_tests.sources.declarative.parsers.testing_components.TestingSomeComponent",
+    }
+
+    component = factory.create_component(CustomErrorHandlerModel, manifest, input_config)
+
+    assert isinstance(component, TestingSomeComponent)
 
 
 def test_custom_components_do_not_contain_extra_fields():