script-build-and-framework-tmp/visualization/backend/app/sanitizer.py

from __future__ import annotations

import json
import re
from typing import Any


_SECRET_PATTERNS = (
    re.compile(
        r"(?i)([\"']?authorization[\"']?\s*[:=]\s*[\"']?)(?:(?:bearer|basic)\s+)?([^\s,\"'\]}]+)"
    ),
    re.compile(
        r"(?i)([\"']?(?:api[_-]?key|access[_-]?token|refresh[_-]?token|token|password|secret)[\"']?\s*[:=]\s*[\"']?)([^\s,\"'\]}]+)"
    ),
    re.compile(r"(?i)(mysql(?:\+pymysql)?://[^:/\s]+:)([^@\s]+)(@)"),
    re.compile(r"(?i)((?:postgres(?:ql)?|redis|mongodb(?:\+srv)?|amqp)://[^:/\s]+:)([^@\s]+)(@)"),
)


def redact_text(value: str) -> str:
    result = value
    for pattern in _SECRET_PATTERNS:
        if pattern.groups == 3:
            result = pattern.sub(r"\1***\3", result)
        else:
            result = pattern.sub(r"\1***", result)
    return result


def sanitize(value: Any, *, max_text: int | None = 120_000) -> Any:
    if isinstance(value, str):
        redacted = redact_text(value)
        if max_text is not None and len(redacted) > max_text:
            return redacted[:max_text] + f"\n…[truncated {len(redacted) - max_text} chars]"
        return redacted
    if isinstance(value, dict):
        result = {}
        for key, nested in value.items():
            key_text = str(key)
            if re.search(r"(?i)(authorization|api[_-]?key|access[_-]?token|secret|password|credential|dsn|database[_-]?url|connection[_-]?string|data[_-]?source[_-]?url)", key_text):
                result[key_text] = "***"
            else:
                result[key_text] = sanitize(nested, max_text=max_text)
        return result
    if isinstance(value, (list, tuple)):
        return [sanitize(v, max_text=max_text) for v in value]
    try:
        json.dumps(value)
        return value
    except TypeError:
        return str(value)