"""Sending reports by email through the configured account, with a copy in its Sent folder. SMTP sends the message. When "save to Sent" is on, the same message is then appended over IMAP to the mailbox's Sent folder (found by its \\Sent flag, or by a usual name), so it shows up there like any message sent from a mail program. Every attempt is written to the mail log, whatever the outcome. """ import base64 import hashlib import imaplib import re import smtplib import socket import ssl import time from dataclasses import dataclass from datetime import datetime, timezone from email.message import EmailMessage from email.utils import formataddr, formatdate, make_msgid from cryptography.fernet import Fernet, InvalidToken from sqlalchemy.orm import Session from .audit import record from .config import settings from .models import MailLog, MailSettings, User TIMEOUT = 30 MAX_RECIPIENTS = 25 ADDRESS = re.compile(r"^[^@\s<>,;\"']+@[^@\s<>,;\"']+\.[^@\s<>,;\"']+$") SENT_NAMES = ("sent", "sent items", "sent messages", "sent mail", "enviados", "itens enviados", "mensagens enviadas", "elementos enviados", "correo enviado") class MailError(Exception): def __init__(self, code: str, detail: str = ""): super().__init__(code) self.code = code self.detail = detail[:500] @dataclass class Attachment: name: str mime: str data: bytes # --- stored passwords ------------------------------------------------------------------------- def _fernet() -> Fernet: key = hashlib.sha256(("mail-password:" + settings.secret_key).encode()).digest() return Fernet(base64.urlsafe_b64encode(key)) def encrypt(password: str) -> str: return _fernet().encrypt(password.encode()).decode() if password else "" def decrypt(token: str) -> str: if not token: return "" try: return _fernet().decrypt(token.encode()).decode() except InvalidToken: # SECRET_KEY changed since the password was saved raise MailError("password_unreadable") def get_settings(db: Session) -> MailSettings: cfg = db.get(MailSettings, 1) if cfg is None: cfg = MailSettings(id=1) db.add(cfg) db.flush() return cfg def is_ready(cfg: MailSettings) -> bool: return bool(cfg.enabled and cfg.smtp_host and cfg.from_address) def clean_addresses(values: list[str] | None) -> list[str]: out = [] for raw in values or []: for part in re.split(r"[,;\s]+", raw or ""): part = part.strip() if not part: continue if not ADDRESS.match(part): raise MailError("invalid_address", part) if part.lower() not in (a.lower() for a in out): out.append(part) return out # --- SMTP ------------------------------------------------------------------------------------- def _smtp(cfg: MailSettings) -> smtplib.SMTP: context = ssl.create_default_context() try: if cfg.smtp_security == "ssl": server = smtplib.SMTP_SSL(cfg.smtp_host, cfg.smtp_port, timeout=TIMEOUT, context=context) else: server = smtplib.SMTP(cfg.smtp_host, cfg.smtp_port, timeout=TIMEOUT) server.ehlo() if cfg.smtp_security == "starttls": server.starttls(context=context) server.ehlo() if cfg.smtp_username: server.login(cfg.smtp_username, decrypt(cfg.smtp_password)) return server except MailError: raise except smtplib.SMTPAuthenticationError as exc: raise MailError("smtp_auth", _smtp_text(exc)) except ssl.SSLError as exc: raise MailError("smtp_tls", str(exc)) except smtplib.SMTPNotSupportedError as exc: raise MailError("smtp_tls", str(exc)) except (socket.gaierror, socket.timeout, ConnectionError, OSError) as exc: raise MailError("smtp_connect", f"{cfg.smtp_host}:{cfg.smtp_port} — {exc}") except smtplib.SMTPException as exc: raise MailError("smtp_error", _smtp_text(exc)) def _smtp_text(exc: Exception) -> str: if isinstance(exc, smtplib.SMTPResponseException): msg = exc.smtp_error.decode(errors="replace") if isinstance(exc.smtp_error, bytes) else str(exc.smtp_error) return f"{exc.smtp_code} {msg}" return str(exc) def smtp_send(cfg: MailSettings, msg: EmailMessage, recipients: list[str]) -> None: server = _smtp(cfg) try: refused = server.send_message(msg, from_addr=cfg.from_address, to_addrs=recipients) if refused: raise MailError("smtp_recipients", ", ".join(refused)) except smtplib.SMTPRecipientsRefused as exc: raise MailError("smtp_recipients", ", ".join(exc.recipients)) except smtplib.SMTPException as exc: raise MailError("smtp_error", _smtp_text(exc)) except OSError as exc: raise MailError("smtp_connect", str(exc)) finally: try: server.quit() except Exception: pass # --- IMAP (copy in the Sent folder) ----------------------------------------------------------- def _imap(cfg: MailSettings) -> imaplib.IMAP4: host = cfg.imap_host or cfg.smtp_host username = cfg.smtp_username if cfg.imap_same_login else cfg.imap_username password = cfg.smtp_password if cfg.imap_same_login else cfg.imap_password context = ssl.create_default_context() try: if cfg.imap_security == "ssl": box = imaplib.IMAP4_SSL(host, cfg.imap_port, ssl_context=context, timeout=TIMEOUT) else: box = imaplib.IMAP4(host, cfg.imap_port, timeout=TIMEOUT) if cfg.imap_security == "starttls": box.starttls(ssl_context=context) box.login(username, decrypt(password)) return box except MailError: raise except imaplib.IMAP4.error as exc: raise MailError("imap_auth", _imap_text(exc)) except ssl.SSLError as exc: raise MailError("imap_tls", str(exc)) except (socket.gaierror, socket.timeout, ConnectionError, OSError) as exc: raise MailError("imap_connect", f"{host}:{cfg.imap_port} — {exc}") def _imap_text(exc: Exception) -> str: text = exc.args[0] if exc.args else str(exc) return text.decode(errors="replace") if isinstance(text, bytes) else str(text) _LIST_LINE = re.compile(r'^\((?P[^)]*)\)\s+(?P"(?:[^"\\]|\\.)*"|NIL)\s+(?P.+)$') def _unquote(text: str) -> str: text = text.strip() if len(text) >= 2 and text[0] == text[-1] == '"': return re.sub(r"\\(.)", r"\1", text[1:-1]) return text def list_folders(box: imaplib.IMAP4) -> list[tuple[str, str, set[str]]]: """(name, delimiter, flags) of every folder. Names stay in IMAP's own encoding for APPEND.""" typ, data = box.list() if typ != "OK": raise MailError("imap_error", "LIST failed") folders = [] for item in data: if isinstance(item, tuple): # name sent as a literal: b'(... ) "/" {12}', b'Itens Enviad' head, literal = item line = head.decode(errors="replace") m = _LIST_LINE.match(re.sub(r"\{\d+\}$", '""', line)) if m: folders.append((literal.decode(errors="replace"), _unquote(m["delim"]), set(m["flags"].lower().split()))) continue if not item: continue m = _LIST_LINE.match(item.decode(errors="replace")) if m: delim = "" if m["delim"] == "NIL" else _unquote(m["delim"]) folders.append((_unquote(m["name"]), delim, set(m["flags"].lower().split()))) return folders def find_sent_folder(box: imaplib.IMAP4, configured: str) -> str: folders = list_folders(box) if configured: if any(name == configured for name, _, _ in folders): return configured raise MailError("sent_folder_not_found", configured) for name, _, flags in folders: if "\\sent" in flags: return name for name, delim, _ in folders: leaf = name.split(delim)[-1] if delim else name if leaf.lower() in SENT_NAMES: return name raise MailError("sent_folder_not_found", ", ".join(name for name, _, _ in folders)[:400]) def _quote(name: str) -> str: return '"' + name.replace("\\", "\\\\").replace('"', '\\"') + '"' def save_copy(cfg: MailSettings, msg: EmailMessage) -> str: box = _imap(cfg) try: folder = find_sent_folder(box, cfg.sent_folder.strip()) typ, data = box.append(_quote(folder), r"(\Seen)", imaplib.Time2Internaldate(time.time()), msg.as_bytes()) if typ != "OK": raise MailError("imap_error", _imap_text(Exception(*(data or [b"APPEND failed"])))) return folder except imaplib.IMAP4.error as exc: raise MailError("imap_error", _imap_text(exc)) finally: try: box.logout() except Exception: pass # --- the whole send --------------------------------------------------------------------------- def build_message(cfg: MailSettings, to: list[str], cc: list[str], subject: str, body: str, attachments: list[Attachment]) -> EmailMessage: msg = EmailMessage() msg["From"] = formataddr((cfg.from_name or "", cfg.from_address)) msg["To"] = ", ".join(to) if cc: msg["Cc"] = ", ".join(cc) if cfg.reply_to: msg["Reply-To"] = cfg.reply_to msg["Subject"] = " ".join(subject.split()) msg["Date"] = formatdate(localtime=True) msg["Message-ID"] = make_msgid(domain=cfg.from_address.split("@")[-1]) text = body.strip() if cfg.signature.strip(): text += "\n\n-- \n" + cfg.signature.strip() msg.set_content(text + "\n") for a in attachments: maintype, subtype = a.mime.split(";")[0].split("/", 1) msg.add_attachment(a.data, maintype=maintype, subtype=subtype, filename=a.name) return msg def send(db: Session, user: User, cfg: MailSettings, *, to: list[str], cc: list[str], subject: str, body: str, attachments: list[Attachment], kind: str = "report", year_id: int | None = None, context: str = "") -> MailLog: """Send, save the Sent copy, and log it. Never raises for delivery problems: the log says what happened.""" log = MailLog(at=datetime.now(timezone.utc), user_id=user.id, kind=kind, year_id=year_id, context=context[:255], sender=cfg.from_address, recipients={"to": to, "cc": cc}, subject=subject[:500], body=body, attachments=[{"name": a.name, "size": len(a.data)} for a in attachments], status="failed", error="", sent_copy="off", sent_copy_detail="", message_id="") try: msg = build_message(cfg, to, cc, subject, body, attachments) log.message_id = msg["Message-ID"] smtp_send(cfg, msg, to + cc) log.status = "sent" except MailError as exc: log.error = (exc.code + (f": {exc.detail}" if exc.detail else ""))[:1000] except (ValueError, TypeError) as exc: log.error = f"invalid_message: {exc}"[:1000] if log.status == "sent" and cfg.save_to_sent: try: folder = save_copy(cfg, msg) log.sent_copy, log.sent_copy_detail = "saved", folder[:500] except MailError as exc: log.sent_copy = "failed" log.sent_copy_detail = (exc.code + (f": {exc.detail}" if exc.detail else ""))[:500] db.add(log) db.flush() where = ", ".join(to + cc) record(db, user, "mail", log.id, "send" if log.status == "sent" else "send_failed", f"{subject} → {where}"[:500], year_id=year_id) db.commit() return log def check(cfg: MailSettings) -> dict: """Sign in to SMTP (and IMAP when the Sent copy is on) without sending anything.""" result = {"smtp": "ok", "imap": "off", "sent_folder": None} try: server = _smtp(cfg) server.quit() except MailError as exc: result["smtp"] = exc.code + (f": {exc.detail}" if exc.detail else "") if cfg.save_to_sent: try: box = _imap(cfg) try: result["sent_folder"] = find_sent_folder(box, cfg.sent_folder.strip()) result["imap"] = "ok" finally: try: box.logout() except Exception: pass except MailError as exc: result["imap"] = exc.code + (f": {exc.detail}" if exc.detail else "") return result