from __future__ import annotations import copy import logging import os from yuxi.channel.secrets.models import is_secret_field_name, is_secret_ref logger = logging.getLogger(__name__) def scrub_env(config: dict, env_prefix_map: dict[str, str]) -> list[str]: """Detect and report plaintext secrets that should be migrated to SecretRef. Returns a list of suggestions like 'config.app_secret → $secret{env:FEISHU_APP_SECRET}'. """ suggestions: list[str] = [] def _walk(prefix: str, d: dict): for key, value in d.items(): path = f"{prefix}.{key}" if prefix else key if isinstance(value, dict): if is_secret_ref(value): continue _walk(path, value) elif is_secret_field_name(key) and isinstance(value, str) and value and value != "***": env_var = env_prefix_map.get(path) if env_var is None: env_var = _derive_env_name(path) logger.debug("Derived env name '%s' for path '%s' via heuristic", env_var, path) suggestions.append(f"{path} → $secret{{env:{env_var}}}") _walk("", config) return suggestions def _derive_env_name(path: str) -> str: return path.replace(".", "_").replace("-", "_").upper() def scrub_plaintext_secrets(config: dict) -> dict: """Remove plaintext secret values from a config dict that have corresponding SecretRef entries. Walks the config and for any secret field that has a plaintext value AND an adjacent SecretRef entry for the same env var, replaces the plaintext with empty string. Returns a deep copy; the original config dict is not modified. """ cleaned = copy.deepcopy(config) def _walk(d: dict, path_prefix: str = ""): for key, value in list(d.items()): current_path = f"{path_prefix}.{key}" if path_prefix else key if isinstance(value, dict): if is_secret_ref(value): continue _walk(value, current_path) elif is_secret_field_name(key): env_key = _derive_env_name(current_path) env_val = os.getenv(env_key) if env_val and value == env_val: d[key] = "" logger.info("Scrubbed plaintext secret at '%s' (matched env %s)", current_path, env_key) _walk(cleaned) return cleaned