DataOps-platform/app/core/data_source/redaction.py

"""Central redaction helpers for data-source requests, logs and responses."""

import re
from collections.abc import Mapping
from urllib.parse import urlsplit, urlunsplit


REDACTED = "[redacted]"
SENSITIVE_KEYS = {
    "password",
    "passwd",
    "credential",
    "credentials",
    "credential_ref",
    "api_key",
    "apikey",
    "token",
    "authorization",
    "encrypted_payload",
    "nonce",
    "conn_str",
    "connection_string",
    "connection_url",
}
RESPONSE_SECRET_KEYS = SENSITIVE_KEYS | {"username"}
_URL_IN_TEXT = re.compile(
    r"(?P<scheme>[a-zA-Z][a-zA-Z0-9+.-]*://)"
    r"(?P<userinfo>[^\s/@]+)@"
)
_NAMED_SECRET = re.compile(
    r"(?i)(password|passwd|api[_-]?key|token|authorization)"
    r"(\s*[=:]\s*)([^\s,;]+)"
)


def _redact_url(value: str) -> str:
    try:
        parsed = urlsplit(value)
    except (TypeError, ValueError):
        return value
    if not parsed.scheme or parsed.hostname is None or "@" not in parsed.netloc:
        return _URL_IN_TEXT.sub(r"\g<scheme>[redacted]@", value)

    host = parsed.hostname
    if ":" in host and not host.startswith("["):
        host = f"[{host}]"
    port = f":{parsed.port}" if parsed.port is not None else ""
    return urlunsplit(
        (
            parsed.scheme,
            f"{REDACTED}@{host}{port}",
            parsed.path,
            parsed.query,
            parsed.fragment,
        )
    )


def redact_mapping(value):
    """Return a recursively redacted copy suitable for logs."""
    if isinstance(value, Mapping):
        redacted = {}
        for key, item in value.items():
            normalized = str(key).strip().lower()
            redacted[key] = (
                REDACTED
                if normalized in SENSITIVE_KEYS
                else redact_mapping(item)
            )
        return redacted
    if isinstance(value, list):
        return [redact_mapping(item) for item in value]
    if isinstance(value, tuple):
        return tuple(redact_mapping(item) for item in value)
    if isinstance(value, str):
        return _redact_url(value)
    return value


def strip_sensitive_fields(value):
    """Remove secret field names and values from API response structures."""
    if isinstance(value, Mapping):
        return {
            key: strip_sensitive_fields(item)
            for key, item in value.items()
            if str(key).strip().lower() not in RESPONSE_SECRET_KEYS
        }
    if isinstance(value, list):
        return [strip_sensitive_fields(item) for item in value]
    if isinstance(value, tuple):
        return tuple(strip_sensitive_fields(item) for item in value)
    if isinstance(value, str):
        return _redact_url(value)
    return value


def sanitize_exception(error: BaseException, limit: int = 1000) -> str:
    """Return a bounded error classification with credentials removed."""
    value = str(error)
    value = _URL_IN_TEXT.sub(r"\g<scheme>[redacted]@", value)
    value = _NAMED_SECRET.sub(
        lambda match: f"{match.group(1)}{match.group(2)}{REDACTED}",
        value,
    )
    return value[: max(0, int(limit))]