7743919695
Bot de Telegram para monitoreo de notificaciones judiciales del PJ Paraguay. Multi-tenant con onboarding conversacional, audit log, y polling automático. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
67 lines
2.3 KiB
Python
67 lines
2.3 KiB
Python
"""
|
|
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
|