diff --git a/airbyte_cdk/cli/airbyte_cdk/_secrets.py b/airbyte_cdk/cli/airbyte_cdk/_secrets.py index 1fb689c8e..4f68b4b2d 100644 --- a/airbyte_cdk/cli/airbyte_cdk/_secrets.py +++ b/airbyte_cdk/cli/airbyte_cdk/_secrets.py @@ -459,25 +459,60 @@ def _print_ci_secrets_masks( _print_ci_secrets_masks_for_config(config=config_dict) +def _print_ci_secret_mask_for_string(secret: str) -> None: + """Print GitHub CI mask for a single secret string. + + We expect single-line secrets, but we also handle the case where the secret contains newlines. + For multi-line secrets, we must print a secret mask for each line separately. + """ + for line in secret.splitlines(): + if line.strip(): # Skip empty lines + print(f"::add-mask::{line!s}") + + +def _print_ci_secret_mask_for_value(value: Any) -> None: + """Print GitHub CI mask for a single secret value. + + Call this function for any values identified as secrets, regardless of type. + """ + if isinstance(value, dict): + # For nested dicts, we call recursively on each value + for v in value.values(): + _print_ci_secret_mask_for_value(v) + + return + + if isinstance(value, list): + # For lists, we call recursively on each list item + for list_item in value: + _print_ci_secret_mask_for_value(list_item) + + return + + # For any other types besides dict and list, we convert to string and mask each line + # separately to handle multi-line secrets (e.g. private keys). + for line in str(value).splitlines(): + if line.strip(): # Skip empty lines + _print_ci_secret_mask_for_string(line) + + def _print_ci_secrets_masks_for_config( config: dict[str, str] | list[Any] | Any, ) -> None: """Print GitHub CI mask for secrets config, navigating child nodes recursively.""" if isinstance(config, list): + # Check each item in the list to look for nested dicts that may contain secrets: for item in config: _print_ci_secrets_masks_for_config(item) - if isinstance(config, dict): + elif isinstance(config, dict): for key, value in config.items(): if _is_secret_property(key): logger.debug(f"Masking secret for config key: {key}") - print(f"::add-mask::{value!s}") - if isinstance(value, dict): - # For nested dicts, we also need to mask the json-stringified version - print(f"::add-mask::{json.dumps(value)!s}") - - if isinstance(value, (dict, list)): - _print_ci_secrets_masks_for_config(config=value) + _print_ci_secret_mask_for_value(value) + elif isinstance(value, (dict, list)): + # Recursively check nested dicts and lists + _print_ci_secrets_masks_for_config(value) def _is_secret_property(property_name: str) -> bool: diff --git a/unit_tests/cli/airbyte_cdk/test_secret_masks.py b/unit_tests/cli/airbyte_cdk/test_secret_masks.py new file mode 100644 index 000000000..72f31c06a --- /dev/null +++ b/unit_tests/cli/airbyte_cdk/test_secret_masks.py @@ -0,0 +1,42 @@ +# Copyright (c) 2025 Airbyte, Inc., all rights reserved. +"""Unit tests for the secret masking functionality in the Airbyte CDK CLI.""" + +from unittest.mock import patch + +import pytest + +from airbyte_cdk.cli.airbyte_cdk import _secrets + + +@pytest.mark.parametrize( + "config,expected_calls", + [ + # Test masking in a flat dict + ({"password": "secret123", "regular": "value"}, ["secret123"]), + # Test masking in a nested dict + ({"outer": {"api_key": "keyval"}}, ["keyval"]), + # Test masking in a list of dicts + ([{"token": "tok1"}, {"name": "v"}], ["tok1"]), + # Test masking in a dict with a list value + ({"passwords": ["a", "b"]}, ["a", "b"]), + # Test masking of multi-line secrets + ({"password": "multi\nline\nsecret"}, ["multi", "line", "secret"]), + # Test masking in a deeply nested structure + ({"a": [{"b": {"secret": "deep"}}]}, ["deep"]), + # Test masking with no secrets + ({"foo": "bar"}, []), + # Additional edge case: mixed types + ({"password": ["a", 123, {"nested": "val"}]}, ["a", "123", "val"]), + ([{"password": "foo"}], ["foo"]), + ], +) +def test_print_ci_secrets_masks_for_config( + config: dict, + expected_calls: list, +) -> None: + with patch( + "airbyte_cdk.cli.airbyte_cdk._secrets._print_ci_secret_mask_for_string", + ) as mask_mock: + _secrets._print_ci_secrets_masks_for_config(config) + actual_calls = [str(call.args[0]) for call in mask_mock.call_args_list] + assert sorted(actual_calls) == sorted(expected_calls)