Skip to content
Merged
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
41 changes: 33 additions & 8 deletions airbyte_cdk/cli/airbyte_cdk/_secrets.py
Original file line number Diff line number Diff line change
Expand Up @@ -459,25 +459,50 @@
_print_ci_secrets_masks_for_config(config=config_dict)


def _print_ci_secret_mask_for_value(value: Any) -> None:
"""Print GitHub CI mask for a single secret value.
Dict and list values masked as their JSON stringified versions.
"""
if isinstance(value, dict):
# For nested dicts, we also need to mask the json-stringified version
for v in value.values():
_print_ci_secret_mask_for_value(v)

return

if isinstance(value, list):
# For lists, we also need to mask the json-stringified version
for list_item in value:
_print_ci_secret_mask_for_value(list_item)

return

# For other types besides dict and list, we convert to string and mask each line
# separately to handle multi-line secrets (e.g. private keys)

secret_str_lines = str(value).splitlines()
for line in secret_str_lines:
print(f"::add-mask::{line!s}")


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:
Expand Down
Loading