nessi
5ed613d441
- Change default use_tls from True to False to match typical STARTTLS setup - Add MailDeliveryError exception for mail delivery failures - Wrap send_mail calls in try-catch blocks to handle errors gracefully - Return 502 status code with error details when mail delivery fails - Add SMTP security mode selector in frontend (STARTTLS/TLS/None) - Add test mail form to admin panel - Handle empty SMTP credentials properly in update_mail_settings - Catch
79 lines
2.9 KiB
Python
79 lines
2.9 KiB
Python
import smtplib
|
|
import socket
|
|
from email.message import EmailMessage
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.config import get_settings
|
|
from app.core.security import decrypt_secret, encrypt_secret
|
|
from app.models.entities import MailSetting
|
|
from app.schemas.common import MailSettingsIn, MailSettingsOut
|
|
|
|
|
|
class MailDeliveryError(RuntimeError):
|
|
pass
|
|
|
|
|
|
def get_mail_settings(db: Session) -> MailSetting:
|
|
settings = db.get(MailSetting, 1)
|
|
if not settings:
|
|
settings = MailSetting(id=1)
|
|
db.add(settings)
|
|
db.flush()
|
|
return settings
|
|
|
|
|
|
def serialize_mail_settings(settings: MailSetting) -> MailSettingsOut:
|
|
return MailSettingsOut(
|
|
smtp_host=settings.smtp_host,
|
|
smtp_port=settings.smtp_port,
|
|
smtp_user=settings.smtp_user,
|
|
has_password=bool(settings.smtp_password_encrypted),
|
|
use_tls=settings.use_tls,
|
|
use_starttls=settings.use_starttls,
|
|
sender_address=settings.sender_address,
|
|
sender_name=settings.sender_name,
|
|
)
|
|
|
|
|
|
def update_mail_settings(db: Session, payload: MailSettingsIn) -> MailSetting:
|
|
settings = get_mail_settings(db)
|
|
settings.smtp_host = payload.smtp_host or None
|
|
settings.smtp_port = payload.smtp_port
|
|
settings.smtp_user = payload.smtp_user or None
|
|
if payload.smtp_password:
|
|
settings.smtp_password_encrypted = encrypt_secret(payload.smtp_password)
|
|
settings.use_tls = payload.use_tls and not payload.use_starttls
|
|
settings.use_starttls = payload.use_starttls
|
|
settings.sender_address = str(payload.sender_address) if payload.sender_address else None
|
|
settings.sender_name = payload.sender_name
|
|
return settings
|
|
|
|
|
|
def send_mail(db: Session, to: str, subject: str, body: str) -> None:
|
|
settings = get_mail_settings(db)
|
|
if not settings.smtp_host or not settings.sender_address:
|
|
raise MailDeliveryError("SMTP is not configured")
|
|
message = EmailMessage()
|
|
message["From"] = f"{settings.sender_name} <{settings.sender_address}>"
|
|
message["To"] = to
|
|
message["Subject"] = subject
|
|
message.set_content(body)
|
|
password = decrypt_secret(settings.smtp_password_encrypted)
|
|
client_cls = smtplib.SMTP_SSL if settings.use_tls and not settings.use_starttls else smtplib.SMTP
|
|
try:
|
|
with client_cls(settings.smtp_host, settings.smtp_port, timeout=20) as smtp:
|
|
if settings.use_starttls:
|
|
smtp.starttls()
|
|
if settings.smtp_user and password:
|
|
smtp.login(settings.smtp_user, password)
|
|
smtp.send_message(message)
|
|
except (OSError, smtplib.SMTPException, socket.timeout) as exc:
|
|
raise MailDeliveryError(f"SMTP delivery failed: {exc}") from exc
|
|
|
|
|
|
def invite_body(token: str) -> str:
|
|
link = f"{get_settings().instance_url.rstrip('/')}/accept-invite?token={token}"
|
|
return f"Welcome to NexaPantry.\n\nOpen this invitation link to set your password:\n{link}\n\nThe link expires automatically."
|
|
|