fix: restore _is_sensitive_field and sensitive field redaction in _safe_deepcopy_config

The function and associated constants (_SENSITIVE_FIELDS_EXACT,
_SENSITIVE_SUFFIXES) were accidentally dropped during the merge.
Restores telemetry-safe config cloning that redacts API keys,
passwords, and other secrets while preserving runtime auth objects.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Soumil Rathi
2026-04-13 10:53:56 -07:00
parent 59880a6f8f
commit 573ca54637
+62 -11
View File
@@ -54,6 +54,8 @@ logger = logging.getLogger(__name__)
# Fields that hold runtime auth/connection objects and must be preserved.
# These are non-serializable objects (e.g. AWSV4SignerAuth, RequestsHttpConnection)
# needed by clients like OpenSearch — not sensitive strings to redact.
_RUNTIME_FIELDS = frozenset({
"http_auth",
"auth",
@@ -61,21 +63,70 @@ _RUNTIME_FIELDS = frozenset({
"ssl_context",
})
# Fields that are known to contain sensitive secrets and must be redacted.
_SENSITIVE_FIELDS_EXACT = frozenset({
"api_key",
"secret_key",
"private_key",
"access_key",
"password",
})
# Suffixes that indicate a field likely holds a secret value.
_SENSITIVE_SUFFIXES = (
"_password",
"_secret",
"_token",
"_credential",
"_credentials",
)
def _is_sensitive_field(field_name: str) -> bool:
"""Check if a field should be redacted for telemetry safety.
Uses a layered approach:
1. Runtime fields (allowlist) — always preserved, highest priority.
2. Exact deny list — known secret field names.
3. Suffix deny list — catches patterns like db_password, auth_secret, etc.
"""
name = field_name.lower().strip()
if name in _RUNTIME_FIELDS:
return False
if name in _SENSITIVE_FIELDS_EXACT:
return True
return any(name.endswith(suffix) for suffix in _SENSITIVE_SUFFIXES)
def _safe_deepcopy_config(config):
"""Deep copy a config object, falling back to shallow copy if needed."""
"""Safely deepcopy config, falling back to dict-based cloning for non-serializable objects."""
try:
copied = deepcopy(config)
except Exception:
copied = config.__class__(**{k: v for k, v in config.__dict__.items()})
# Restore non-serializable runtime fields from the original config
for field in _RUNTIME_FIELDS:
if hasattr(config, field):
return deepcopy(config)
except Exception as e:
logger.debug(f"Deepcopy failed, using dict-based cloning: {e}")
config_class = type(config)
if hasattr(config, "model_dump"):
try:
setattr(copied, field, getattr(config, field))
except (AttributeError, TypeError):
pass
return copied
clone_dict = config.model_dump()
except Exception:
clone_dict = dict(config.__dict__)
else:
clone_dict = dict(config.__dict__)
# Restore runtime fields, redact sensitive ones
for field_name in list(clone_dict.keys()):
if field_name in _RUNTIME_FIELDS and hasattr(config, field_name):
clone_dict[field_name] = getattr(config, field_name)
elif _is_sensitive_field(field_name):
clone_dict[field_name] = None
try:
return config_class(**clone_dict)
except Exception:
logger.debug("Config reconstruction failed, returning shallow dict clone")
return type("Config", (), clone_dict)()
def _normalize_iso_timestamp_to_utc(timestamp: Optional[str]) -> Optional[str]: