""" utils/sanitize.py — Sanitización de datos antes de loggear. REGLA: todo log que incluya datos externos (responses del PJ, inputs del usuario) debe pasar por credential_scrubber() antes de escribirse. Uso: from utils.sanitize import scrub logger.info("respuesta_pj", data=scrub(response_text)) logger.error("error_en_request", body=scrub(request_body)) """ from __future__ import annotations import re # Patrones que NUNCA deben aparecer en logs _PATTERNS = [ # Bearer tokens (opacos base64url, ~128 chars) (re.compile(r"Bearer\s+[A-Za-z0-9\-_]{20,}"), "Bearer [REDACTED]"), # Authorization header completo (re.compile(r'"[Aa]uthorization"\s*:\s*"[^"]+"'), '"Authorization": "[REDACTED]"'), # bearerToken en JSON (re.compile(r'"bearerToken"\s*:\s*"[^"]+"'), '"bearerToken": "[REDACTED]"'), # refreshToken en JSON (re.compile(r'"refreshToken"\s*:\s*"[^"]+"'), '"refreshToken": "[REDACTED]"'), # permisoFirmado JWT (eyJ...) (re.compile(r'"permisoFirmado"\s*:\s*"eyJ[A-Za-z0-9\.\-_=+/]+"'), '"permisoFirmado": "[REDACTED]"'), # clave/password en JSON (campo "clave" del body de login) (re.compile(r'"clave"\s*:\s*"[^"]+"'), '"clave": "[REDACTED]"'), (re.compile(r'"password"\s*:\s*"[^"]+"'), '"password": "[REDACTED]"'), # cookiesession (hexadecimal 32 chars) (re.compile(r"cookiesession1=[A-Fa-f0-9]{32}"), "cookiesession1=[REDACTED]"), # Cualquier JWT genérico (eyJ...) (re.compile(r"\beyJ[A-Za-z0-9\.\-_=+/]{50,}"), "[JWT_REDACTED]"), ] def scrub(value: object) -> str: """ Convierte value a string y aplica todos los patrones de sanitización. Usar siempre antes de pasar datos externos a structlog. Ejemplo: logger.debug("raw_response", body=scrub(response.text)) """ text = str(value) for pattern, replacement in _PATTERNS: text = pattern.sub(replacement, text) return text def scrub_dict(d: dict) -> dict: """ Aplica scrub() recursivamente a todos los valores string de un dict. Útil para sanitizar dicts de headers o payloads antes de loggear. """ result = {} for key, value in d.items(): if isinstance(value, str): result[key] = scrub(value) elif isinstance(value, dict): result[key] = scrub_dict(value) else: result[key] = value return result