""" Validadores del contenido del correo. El endpoint original pasaba los datos del cliente directamente a las cabeceras del mensaje: message['To'] = mail_data['mail_send_to'] message['Subject'] = mail_data['mail_subject'] Sin validar. Un asunto como "Factura\\r\\nBcc: atacante@ejemplo.com" anade una cabecera `Bcc` real al mensaje: cualquiera con una credencial valida podia recibir copia oculta de todos los correos, o falsear el remitente. Es inyeccion de cabeceras SMTP, y se corta aqui rechazando cualquier caracter de control en los campos que acaban siendo cabeceras. """ from __future__ import annotations import logging import mimetypes import os import re import unicodedata from django.conf import settings from rest_framework import serializers logger = logging.getLogger("notifications.validation") # Caracteres de control: los que permiten partir una cabecera en dos. Se # rechazan todos, no solo \r y \n, porque los analizadores de cabeceras de # los distintos servidores tratan de forma inconsistente el resto (\x00 corta # la cadena en implementaciones en C, \x0b y \x0c cuentan como salto de linea # en algunos analizadores). _CONTROL_CHARACTERS = re.compile(r"[\x00-\x1f\x7f]") # Identificador valido para una imagen incrustada (`cid:`). _CONTENT_ID = re.compile(r"^[A-Za-z0-9._-]{1,64}$") # Nombre de cabecera valido segun RFC 5322 (sin dos puntos ni espacios). _HEADER_NAME = re.compile(r"^[A-Za-z0-9-]{1,64}$") # Cabeceras que fija la propia API. Permitir que el cliente las sobrescriba # equivaldria a dejar abierta la inyeccion que se acaba de cerrar. RESERVED_HEADERS = frozenset( { "bcc", "cc", "content-transfer-encoding", "content-type", "date", "from", "message-id", "mime-version", "received", "reply-to", "return-path", "sender", "subject", "to", } ) # --------------------------------------------------------------------------- # Texto de cabeceras # --------------------------------------------------------------------------- def reject_control_characters(value: str, field_name: str = "campo") -> str: """Rechaza el valor si contiene caracteres de control.""" if value and _CONTROL_CHARACTERS.search(value): raise serializers.ValidationError( f"El {field_name} no puede contener saltos de linea ni caracteres " "de control." ) return value class HeaderSafeCharField(serializers.CharField): """`CharField` que ademas prohibe caracteres de control.""" def to_internal_value(self, data): value = super().to_internal_value(data) return reject_control_characters(value, self.field_name or "campo") class HeaderSafeEmailField(serializers.EmailField): """ `EmailField` que prohibe caracteres de control y normaliza el dominio. El validador de DRF ya rechaza la mayoria de las formas invalidas, pero la comprobacion explicita deja constancia de la intencion y protege ante cambios futuros en ese validador. """ def to_internal_value(self, data): value = super().to_internal_value(data) reject_control_characters(value, "correo electronico") local, _, domain = value.rpartition("@") return f"{local}@{domain.lower()}" # --------------------------------------------------------------------------- # Remitente # --------------------------------------------------------------------------- def validate_sender(from_email: str, api_client=None) -> str: """ Comprueba que la API tenga permiso para enviar desde esa direccion. Sin esta comprobacion, cualquier cliente con una credencial valida puede enviar correo firmado por el dominio de la empresa haciendose pasar por quien quiera (`direccion@bancoajeno.com`). Como los mensajes salen con la reputacion e infraestructura del dominio propio, eso convierte la API en una plataforma de suplantacion. La politica se define en `EMAIL_ALLOWED_FROM_DOMAINS` y `EMAIL_ALLOWED_FROM_ADDRESSES`, y cada cliente puede tener una lista mas restrictiva en `allowed_senders`. """ address = from_email.lower() domain = address.rpartition("@")[2] # Politica propia del cliente: si la tiene, es la unica que aplica. client_senders = list(getattr(api_client, "allowed_senders", None) or []) if client_senders: if not _matches(address, domain, client_senders): raise serializers.ValidationError( f"Este cliente no tiene autorizado enviar desde '{from_email}'." ) return from_email allowed_addresses = settings.EMAIL_ALLOWED_FROM_ADDRESSES allowed_domains = settings.EMAIL_ALLOWED_FROM_DOMAINS if not allowed_addresses and not allowed_domains: raise serializers.ValidationError( "No hay remitentes autorizados configurados. Define " "EMAIL_ALLOWED_FROM_DOMAINS o EMAIL_ALLOWED_FROM_ADDRESSES." ) if address in allowed_addresses or domain in allowed_domains: return from_email raise serializers.ValidationError( f"El remitente '{from_email}' no esta en la lista de direcciones " "autorizadas." ) def _matches(address: str, domain: str, patterns: list[str]) -> bool: for pattern in patterns: pattern = pattern.strip().lower() if not pattern: continue if pattern.startswith("@"): if domain == pattern[1:]: return True elif pattern == address or domain == pattern: return True return False # --------------------------------------------------------------------------- # Adjuntos # --------------------------------------------------------------------------- try: # pragma: no cover - depende de la libreria del sistema import magic _MAGIC_AVAILABLE = True except Exception: # noqa: BLE001 - ImportError, OSError segun la plataforma magic = None # type: ignore[assignment] _MAGIC_AVAILABLE = False logger.warning( "libmagic no esta disponible: el tipo de los adjuntos se deducira de la " "extension. Instala libmagic para validar el contenido real." ) def detect_mime_type(content: bytes, filename: str) -> str: """ Determina el tipo MIME real del archivo. Se inspecciona el contenido, no el `Content-Type` que declara el cliente: ese valor lo elige quien sube el archivo y por tanto no vale para decidir si el archivo es aceptable. """ if _MAGIC_AVAILABLE and content: try: return magic.from_buffer(content[:8192], mime=True) or "" except Exception as exc: # noqa: BLE001 - libmagic mal instalada logger.warning("Fallo la deteccion MIME con libmagic: %s", exc) return mimetypes.guess_type(filename)[0] or "application/octet-stream" def sanitize_filename(filename: str) -> str: """ Deja el nombre del archivo en algo seguro de escribir y de mostrar. Se queda solo con el nombre base para que un valor como `../../etc/passwd` no pueda escapar de un directorio si el destinatario guarda el adjunto sin comprobarlo. """ name = unicodedata.normalize("NFKC", filename or "") name = name.replace("\\", "/").split("/")[-1] name = _CONTROL_CHARACTERS.sub("", name).strip().strip(".") if not name: name = "adjunto" return name[:200] def validate_attachment(filename: str, content: bytes) -> str: """ Valida un adjunto y devuelve su tipo MIME. Comprueba tamanio, extension y contenido real. """ size = len(content) if size == 0: raise serializers.ValidationError(f"El adjunto '{filename}' esta vacio.") if size > settings.EMAIL_MAX_ATTACHMENT_BYTES: limit_mb = settings.EMAIL_MAX_ATTACHMENT_BYTES / (1024 * 1024) raise serializers.ValidationError( f"El adjunto '{filename}' supera el limite de {limit_mb:.1f} MB." ) extension = os.path.splitext(filename)[1].lstrip(".").lower() if extension in settings.EMAIL_BLOCKED_ATTACHMENT_EXTENSIONS: raise serializers.ValidationError( f"No se admiten adjuntos con extension '.{extension}'." ) mime_type = detect_mime_type(content, filename) if mime_type not in settings.EMAIL_ALLOWED_ATTACHMENT_MIME_TYPES: raise serializers.ValidationError( f"El adjunto '{filename}' es de tipo '{mime_type}', que no esta " "permitido." ) return mime_type def validate_content_id(content_id: str) -> str: """Valida el identificador de una imagen incrustada.""" if not _CONTENT_ID.match(content_id): raise serializers.ValidationError( f"El content_id '{content_id}' no es valido: solo se admiten " "letras, numeros, punto, guion y guion bajo (maximo 64)." ) return content_id # --------------------------------------------------------------------------- # Cabeceras adicionales # --------------------------------------------------------------------------- def validate_extra_headers(headers: dict) -> dict: """Valida las cabeceras personalizadas que pide el cliente.""" validated: dict[str, str] = {} for raw_name, raw_value in headers.items(): name = str(raw_name).strip() value = str(raw_value) if not _HEADER_NAME.match(name): raise serializers.ValidationError( f"El nombre de cabecera '{name}' no es valido." ) if name.lower() in RESERVED_HEADERS: raise serializers.ValidationError( f"La cabecera '{name}' la gestiona la API y no se puede " "sobrescribir." ) reject_control_characters(value, f"valor de la cabecera '{name}'") if len(value) > 998: # limite de longitud de linea del RFC 5322 raise serializers.ValidationError( f"El valor de la cabecera '{name}' es demasiado largo." ) validated[name] = value return validated