import logging import smtplib import ssl import base64 import httpx import time import threading from email.message import EmailMessage from email.utils import formataddr from app.core.config import get_settings from app.core.db import db_session from app.core.security import decrypt_integration_secret, encrypt_text from sqlalchemy import text log = logging.getLogger(__name__) _token_lock = threading.Lock() _token_cache = {"value": None, "expires": 0.0} def _post_with_retry(url: str, **kwargs): last = None for attempt in range(3): try: response = httpx.post(url, **kwargs) if response.status_code < 500 and response.status_code != 429: return response last = RuntimeError(f"REMOTE_HTTP_{response.status_code}") except httpx.RequestError as exc: last = exc if attempt < 2: time.sleep(1.5 * (2 ** attempt)) raise last or RuntimeError("REMOTE_UNAVAILABLE") def _google_access_token(config) -> str: now = time.monotonic() with _token_lock: if _token_cache["value"] and _token_cache["expires"] > now + 60: return str(_token_cache["value"]) client_secret = decrypt_integration_secret(config["google_client_secret_cifrado"]) refresh_token = decrypt_integration_secret(config["google_refresh_token_cifrado"]) response = _post_with_retry("https://oauth2.googleapis.com/token", data={ "client_id": config["google_client_id"], "client_secret": client_secret, "refresh_token": refresh_token, "grant_type": "refresh_token"}, timeout=20.0) if response.status_code != 200: reason = "TOKEN_REJECTED" try: reason = response.json().get("error", reason) except Exception: pass raise RuntimeError(f"GOOGLE_TOKEN_{response.status_code}_{reason}") payload = response.json() _token_cache["value"] = payload["access_token"] _token_cache["expires"] = now + int(payload.get("expires_in", 3600)) return str(_token_cache["value"]) def send_email(to: str, subject: str, body: str, *, html_body: str | None = None, attachments: list[tuple[str, bytes, str, str]] | None = None, bcc: list[str] | None = None) -> None: settings = get_settings() config = None try: with db_session() as db: config = db.execute(text("SELECT * FROM configuracion_correo WHERE id=1 AND activo=TRUE")).mappings().first() except Exception: log.warning("mail_config_table_unavailable") simulated = settings.mail_mode == "log" and not config host = config["smtp_host"] if config else settings.smtp_host port = int(config["smtp_port"] if config else settings.smtp_port) user = config["smtp_usuario"] if config else settings.smtp_user password = decrypt_integration_secret(config["smtp_password_cifrado"]) if config and config["smtp_password_cifrado"] else settings.smtp_password sender = config["smtp_desde"] if config else settings.smtp_from security = config["seguridad"] if config else "STARTTLS" status, detail = "SIMULADO" if simulated else "ENVIADO", None message = EmailMessage() sender_name = config.get("nombre_remitente") if config else None message["From"] = formataddr((sender_name, sender)) if sender_name else sender message["To"] = to if bcc: message["Bcc"] = ", ".join(dict.fromkeys(x for x in bcc if x and x != to)) message["Subject"] = subject message.set_content(body) if html_body: message.add_alternative(html_body, subtype="html") for filename, content, maintype, subtype in attachments or []: message.add_attachment(content, maintype=maintype, subtype=subtype, filename=filename) try: if simulated: log.info("email_mock to=%s subject=%s", to, subject) elif not host or not user: raise RuntimeError("SMTP no configurado") else: oauth = bool(config and config.get("auth_mode") == "OAUTH2") access_token = None if oauth: access_token = _google_access_token(config) if oauth: raw = base64.urlsafe_b64encode(message.as_bytes()).decode().rstrip("=") send_response = _post_with_retry("https://gmail.googleapis.com/gmail/v1/users/me/messages/send", headers={"Authorization":f"Bearer {access_token}","Content-Type":"application/json"}, json={"raw":raw},timeout=25.0) if send_response.status_code not in (200, 201): reason = "REJECTED" try: reason = send_response.json().get("error",{}).get("status",reason) except Exception: pass raise RuntimeError(f"GMAIL_API_{send_response.status_code}_{reason}") else: smtp_factory = smtplib.SMTP_SSL if security == "SSL" else smtplib.SMTP kwargs = {"host":host,"port":port,"timeout":20} if security == "SSL": kwargs["context"] = ssl.create_default_context() with smtp_factory(**kwargs) as smtp: if security != "SSL": smtp.ehlo(); smtp.starttls(context=ssl.create_default_context()); smtp.ehlo() if password: smtp.login(user, password) else: raise RuntimeError("Credencial SMTP no configurada") smtp.send_message(message) except Exception as exc: safe = str(exc) if isinstance(exc, RuntimeError) else type(exc).__name__ status, detail = "ERROR", safe[:500] raise finally: try: with db_session() as db: db.execute(text("""INSERT INTO correos_auditoria (destinatario_cifrado,asunto,estado,proveedor,detalle_error,enviado_en) VALUES(:to,:subject,:status,:provider,:detail,IF(:status='ENVIADO',UTC_TIMESTAMP(6),NULL))"""), {"to":encrypt_text(to),"subject":subject[:255],"status":status, "provider":"GOOGLE_WORKSPACE" if config else "ENV", "detail":detail}) except Exception: log.exception("mail_audit_write_failed") def notify_admin_error(correlation_id: str, component: str, code: str) -> None: settings = get_settings() send_email(settings.admin_error_email, settings.admin_error_subject, f"Error ID: {correlation_id}\nComponente: {component}\nCódigo: {code}\n" "Revisar logs sanitizados del servidor. No se incluyen datos personales.")