from typing import Any from app.core.config import get_settings MASKED_VALUE = "[MASKED]" SENSITIVE_RESPONSE_KEYS = frozenset( { "account_number", "api_key", "bank_account", "card_no", "direct_llm_api_key", "email", "feishu_app_secret", "feishu_verification_token", "hermes_api_key", "id_card", "mobile", "openclaw_gateway_token", "password", "payment_account", "phone", "secret", "tenant_access_token", "token", } ) def mask_configured(value: Any, domain: str | None = None) -> Any: """Mask sensitive response fields when response masking is enabled.""" settings = get_settings() if not settings.mask_sensitive_responses: return value configured_fields = {item.lower() for item in settings.masked_response_fields} return mask_sensitive(value, domain=domain, configured_fields=configured_fields) def mask_sensitive( value: Any, domain: str | None = None, configured_fields: set[str] | None = None, ) -> Any: if isinstance(value, dict): masked: dict[str, Any] = {} for key, item in value.items(): key_text = str(key) if _should_mask(key_text, domain, configured_fields or set()): masked[key_text] = MASKED_VALUE else: masked[key_text] = mask_sensitive(item, domain, configured_fields) return masked if isinstance(value, list): return [mask_sensitive(item, domain, configured_fields) for item in value] if isinstance(value, tuple): return [mask_sensitive(item, domain, configured_fields) for item in value] return value def _should_mask(key: str, domain: str | None, configured_fields: set[str]) -> bool: field = key.lower() if field in SENSITIVE_RESPONSE_KEYS or field in configured_fields: return True if f"*.{field}" in configured_fields: return True return bool(domain and f"{domain.lower()}.{field}" in configured_fields)