fix(manifest-server): keep caller-supplied manifests gated via custom_components_trusted

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

Author: devin-ai-integration[bot]

airbyte_cdk/manifest_server/command_processor/utils.py

@@ -79,6 +79,9 @@ def build_source(
         state=state,
         source_config=definition,
         config=config,
+        # The manifest is supplied by the API caller rather than bundled in a connector image, so
+        # any custom components it references are untrusted code.
+        custom_components_trusted=False,
         normalize_manifest=should_normalize,
         migrate_manifest=should_migrate,
         emit_connector_builder_messages=True,

airbyte_cdk/sources/declarative/concurrent_declarative_source.py

@@ -150,6 +150,7 @@ def __init__(
         normalize_manifest: bool = False,
         limits: Optional[TestLimits] = None,
         config_path: Optional[str] = None,
+        custom_components_trusted: bool = True,
         **kwargs: Any,
     ) -> None:
         self.logger = logging.getLogger(f"airbyte.{self.name}")
@@ -174,6 +175,7 @@ def __init__(
         # the declarative models into runtime components. Concurrent sources will continue to checkpoint
         # incremental streams running in full refresh.
         component_factory = ModelToComponentFactory(
+            custom_components_trusted=custom_components_trusted,
             emit_connector_builder_messages=emit_connector_builder_messages,
             message_repository=ConcurrentMessageRepository(queue, message_repository),
             configured_catalog=catalog,
@@ -197,6 +199,7 @@ def __init__(
             component_factory
             if component_factory
             else ModelToComponentFactory(
+                custom_components_trusted=custom_components_trusted,
                 emit_connector_builder_messages=emit_connector_builder_messages,
                 max_concurrent_async_job_count=source_config.get("max_concurrent_async_job_count"),
             )

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -705,8 +705,10 @@ def __init__(
         max_concurrent_async_job_count: Optional[int] = None,
         configured_catalog: Optional[ConfiguredAirbyteCatalog] = None,
         api_budget: Optional[APIBudget] = None,
+        custom_components_trusted: bool = True,
     ):
         self._init_mappings()
+        self._custom_components_trusted = custom_components_trusted
         self._limit_pages_fetched_per_slice = limit_pages_fetched_per_slice
         self._limit_slices_fetched = limit_slices_fetched
         self._emit_connector_builder_messages = emit_connector_builder_messages
@@ -1815,11 +1817,15 @@ 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`. 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():
+        # by `class_name`. Manifests supplied by a caller, whether through the config or directly to
+        # the manifest server, are untrusted input and could point `class_name` at any importable
+        # callable, so they honor 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.
+        manifest_is_untrusted = not self._custom_components_trusted or bool(
+            config.get(INJECTED_MANIFEST)
+        )
+        if manifest_is_untrusted and not custom_code_execution_permitted():
             raise AirbyteCustomCodeNotPermittedError
 
         custom_component_class = self._get_class_from_fully_qualified_class_name(model.class_name)
@@ -4141,6 +4147,7 @@ def create_parent_stream_config_with_substream_wrapper(
         )
 
         substream_factory = ModelToComponentFactory(
+            custom_components_trusted=self._custom_components_trusted,
             connector_state_manager=connector_state_manager,
             limit_pages_fetched_per_slice=self._limit_pages_fetched_per_slice,
             limit_slices_fetched=self._limit_slices_fetched,

unit_tests/sources/declarative/parsers/test_model_to_component_factory.py

@@ -2578,6 +2578,34 @@ def test_create_custom_component_permitted_for_bundled_manifest(monkeypatch):
     assert isinstance(component, TestingSomeComponent)
 
 
+def test_create_custom_component_requires_custom_code_enabled_when_untrusted(monkeypatch):
+    """A caller-supplied manifest must honor the gate even when it never passes through the config.
+
+    The manifest server receives the manifest as a request payload rather than through the
+    config, so its untrusted provenance is signalled by `custom_components_trusted=False`.
+    """
+    monkeypatch.delenv(ENV_VAR_ALLOW_CUSTOM_CODE, raising=False)
+
+    untrusted_factory = ModelToComponentFactory(custom_components_trusted=False)
+
+    def _fail_if_resolved(*args, **kwargs):
+        raise AssertionError(
+            "`class_name` must not be resolved or imported when custom code is disabled"
+        )
+
+    monkeypatch.setattr(
+        untrusted_factory, "_get_class_from_fully_qualified_class_name", _fail_if_resolved
+    )
+
+    manifest = {
+        "type": "CustomErrorHandler",
+        "class_name": "unit_tests.sources.declarative.parsers.testing_components.TestingSomeComponent",
+    }
+
+    with pytest.raises(AirbyteCustomCodeNotPermittedError):
+        untrusted_factory.create_component(CustomErrorHandlerModel, manifest, input_config)
+
+
 def test_custom_components_do_not_contain_extra_fields():
     custom_substream_partition_router_manifest = {
         "type": "CustomPartitionRouter",