diff --git a/airbyte_cdk/sources/concurrent_source/concurrent_read_processor.py b/airbyte_cdk/sources/concurrent_source/concurrent_read_processor.py index 160ec1ef03..e045b4ef9c 100644 --- a/airbyte_cdk/sources/concurrent_source/concurrent_read_processor.py +++ b/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) diff --git a/airbyte_cdk/sources/declarative/async_job/job_orchestrator.py b/airbyte_cdk/sources/declarative/async_job/job_orchestrator.py index 3607ad0264..f8c1179c90 100644 --- a/airbyte_cdk/sources/declarative/async_job/job_orchestrator.py +++ b/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 diff --git a/airbyte_cdk/sources/declarative/interpolation/jinja.py b/airbyte_cdk/sources/declarative/interpolation/jinja.py index 415a0c8572..bbb09e47d8 100644 --- a/airbyte_cdk/sources/declarative/interpolation/jinja.py +++ b/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: diff --git a/unit_tests/sources/declarative/async_job/test_job_orchestrator.py b/unit_tests/sources/declarative/async_job/test_job_orchestrator.py index dce40e6242..a482322e17 100644 --- a/unit_tests/sources/declarative/async_job/test_job_orchestrator.py +++ b/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 diff --git a/unit_tests/sources/declarative/interpolation/test_jinja.py b/unit_tests/sources/declarative/interpolation/test_jinja.py index c358fe4aa5..be90c5413a 100644 --- a/unit_tests/sources/declarative/interpolation/test_jinja.py +++ b/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", diff --git a/unit_tests/sources/streams/concurrent/test_concurrent_read_processor.py b/unit_tests/sources/streams/concurrent/test_concurrent_read_processor.py index 7b38486abe..06c08ea72b 100644 --- a/unit_tests/sources/streams/concurrent/test_concurrent_read_processor.py +++ b/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]