fix(low-code): stop leaking interpolation context in undeclared-variable errors

airbyte_cdk/sources/concurrent_source/concurrent_read_processor.py

@@ -250,13 +250,15 @@ def on_exception(self, exception: StreamThreadException) -> Iterable[AirbyteMess
 
         stream_descriptor = StreamDescriptor(name=exception.stream_name)
         if isinstance(exception.exception, AirbyteTracedException):
-            yield exception.exception.as_airbyte_message(stream_descriptor=stream_descriptor)
+            yield exception.exception.as_sanitized_airbyte_message(
+                stream_descriptor=stream_descriptor
+            )
         else:
             yield AirbyteTracedException.from_exception(
                 exception.exception,
                 stream_descriptor=stream_descriptor,
                 message=f"An unexpected error occurred in stream {exception.stream_name}: {type(exception.exception).__name__}",
-            ).as_airbyte_message()
+            ).as_sanitized_airbyte_message()
 
     def _flag_exception(self, stream_name: str, exception: Exception) -> None:
         self._exceptions_per_stream_name.setdefault(stream_name, []).append(exception)

airbyte_cdk/sources/declarative/async_job/job_orchestrator.py

@@ -303,7 +303,7 @@ def _keep_api_budget_with_failed_job(
         )
         # Even though we're not sure this will break the stream, we will emit here for simplicity's sake. If we wanted to be more accurate,
         # we would keep the exceptions in-memory until we know that we have reached the max attempt.
-        self._message_repository.emit_message(traced_exception.as_airbyte_message())
+        self._message_repository.emit_message(traced_exception.as_sanitized_airbyte_message())
         job = self._create_failed_job(_slice)
         self._job_tracker.add_job(intent, job.api_job_id())
         return job

airbyte_cdk/sources/declarative/interpolation/jinja.py

@@ -139,7 +139,8 @@ def _eval(self, s: Optional[str], context: Mapping[str, Any]) -> Optional[str]:
             undeclared_not_in_context = {var for var in undeclared if var not in context}
             if undeclared_not_in_context:
                 raise ValueError(
-                    f"Jinja macro has undeclared variables: {undeclared_not_in_context}. Context: {context}"
+                    f"Jinja macro has undeclared variables: {sorted(undeclared_not_in_context)}. "
+                    f"Template: {s!r}. Available variables: {sorted(context.keys())}."
                 )
             return self._compile(s).render(context)  # type: ignore # from_string is able to handle None
         except TypeError:

unit_tests/sources/declarative/async_job/test_job_orchestrator.py

@@ -22,6 +22,7 @@
 from airbyte_cdk.sources.message import MessageRepository
 from airbyte_cdk.sources.types import StreamSlice
 from airbyte_cdk.utils import AirbyteTracedException
+from airbyte_cdk.utils.airbyte_secrets_utils import update_secrets
 
 _ANY_STREAM_SLICE = Mock()
 _A_STREAM_SLICE = Mock()
@@ -386,6 +387,25 @@ def test_given_start_job_raise_when_create_and_get_completed_partitions_then_fre
 
         assert job_tracker.try_to_get_intent()
 
+    def test_given_start_job_exception_contains_secret_when_start_job_then_emit_sanitized_trace(
+        self,
+    ) -> None:
+        secret = "FAKE_SECRET_TOKEN"
+        update_secrets([secret])
+        try:
+            self._job_repository.start.side_effect = RuntimeError(secret)
+            orchestrator = self._orchestrator([_A_STREAM_SLICE])
+
+            orchestrator._start_job(_A_STREAM_SLICE)
+
+            trace_error = self._message_repository.emit_message.call_args.args[0].trace.error
+            assert secret not in trace_error.message
+            assert secret not in trace_error.internal_message
+            assert secret not in trace_error.stack_trace
+            assert "****" in trace_error.internal_message
+        finally:
+            update_secrets([])
+
     @mock.patch(sleep_mock_target)
     def test_given_skipped_when_create_and_get_completed_partitions_then_skip_without_fetching_records(
         self, mock_sleep: MagicMock

unit_tests/sources/declarative/interpolation/test_jinja.py

@@ -251,6 +251,35 @@ def test_undeclared_variables(template_string, expected_error, expected_value):
         assert actual_value == expected_value
 
 
+@pytest.mark.parametrize(
+    "template, undeclared_variables",
+    [
+        ("{{ components_values['x'] }}", ["components_values"]),
+        ("{{ b_var }} {{ a_var }}", ["a_var", "b_var"]),
+    ],
+)
+def test_undeclared_variables_do_not_include_context_values(template, undeclared_variables):
+    messages = []
+
+    for index in (1, 2):
+        config = {"access_token": f"FAKE_SECRET_TOKEN_{index}"}
+        stream_slice = {"cursor": f"FAKE_SLICE_VALUE_{index}"}
+        with pytest.raises(ValueError) as exc_info:
+            interpolation.eval(template, config, stream_slice=stream_slice)
+
+        message = str(exc_info.value)
+        assert f"FAKE_SECRET_TOKEN_{index}" not in message
+        assert f"FAKE_SLICE_VALUE_{index}" not in message
+        messages.append(message)
+
+    expected_message = (
+        f"Jinja macro has undeclared variables: {undeclared_variables}. "
+        f"Template: {template!r}. Available variables: "
+        "['config', 'stream_interval', 'stream_partition', 'stream_slice']."
+    )
+    assert messages == [expected_message, expected_message]
+
+
 @freeze_time("2021-09-01")
 @pytest.mark.parametrize(
     "template_string, expected_value",

unit_tests/sources/streams/concurrent/test_concurrent_read_processor.py

@@ -50,6 +50,7 @@
 from airbyte_cdk.sources.streams.concurrent.partitions.types import PartitionCompleteSentinel
 from airbyte_cdk.sources.types import Record
 from airbyte_cdk.sources.utils.slice_logger import SliceLogger
+from airbyte_cdk.utils.airbyte_secrets_utils import update_secrets
 from airbyte_cdk.utils.traced_exception import AirbyteTracedException
 
 _STREAM_NAME = "stream"
@@ -825,6 +826,63 @@ def test_on_exception_non_ate_uses_templated_message_with_correct_failure_type(s
         # failure_type defaults to system_error for unhandled exceptions
         assert trace_error.failure_type == FailureType.system_error
 
+    def test_on_exception_sanitizes_airbyte_traced_exception(self):
+        secret = "FAKE_SECRET_TOKEN"
+        update_secrets([secret])
+        try:
+            handler = ConcurrentReadProcessor(
+                [self._stream],
+                self._partition_enqueuer,
+                self._thread_pool_manager,
+                self._logger,
+                self._slice_logger,
+                self._message_repository,
+                self._partition_reader,
+            )
+            traced_exception = AirbyteTracedException(
+                internal_message=secret,
+                message=secret,
+                exception=ValueError(secret),
+            )
+
+            messages = list(
+                handler.on_exception(StreamThreadException(traced_exception, _STREAM_NAME))
+            )
+            trace_error = messages[0].trace.error
+
+            assert secret not in trace_error.message
+            assert secret not in trace_error.internal_message
+            assert secret not in trace_error.stack_trace
+            assert "****" in trace_error.internal_message
+        finally:
+            update_secrets([])
+
+    def test_on_exception_sanitizes_plain_exception(self):
+        secret = "FAKE_SECRET_TOKEN"
+        update_secrets([secret])
+        try:
+            handler = ConcurrentReadProcessor(
+                [self._stream],
+                self._partition_enqueuer,
+                self._thread_pool_manager,
+                self._logger,
+                self._slice_logger,
+                self._message_repository,
+                self._partition_reader,
+            )
+
+            messages = list(
+                handler.on_exception(StreamThreadException(ValueError(secret), _STREAM_NAME))
+            )
+            trace_error = messages[0].trace.error
+
+            assert secret not in trace_error.message
+            assert secret not in trace_error.internal_message
+            assert secret not in trace_error.stack_trace
+            assert "****" in trace_error.internal_message
+        finally:
+            update_secrets([])
+
     @freezegun.freeze_time("2020-01-01T00:00:00")
     def test_start_next_partition_generator(self):
         stream_instances_to_read_from = [self._stream]