335 lines
12 KiB
Python
335 lines
12 KiB
Python
|
"""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<flags>[^)]*)\)\s+(?P<delim>"(?:[^"\\]|\\.)*"|NIL)\s+(?P<name>.+)$')
|
||
|
|
|
||
|
|
|
||
|
|
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
|