import os import smtplib from email.message import EmailMessage def _truthy(v: str | None) -> bool: if v is None: return False return v.strip().lower() in ("1", "true", "yes", "on") def send_email(*, to_email: str, subject: str, body_text: str): """Send a plain-text email using SMTP settings from: 1) Admin-configured settings stored in the database (AppSettings) (highest priority) 2) Environment variables (fallback) If you *do* want environment variables to override DB settings (e.g. in production), set SMTP_OVERRIDE_DB=1. Required configuration (either from DB or env): - host - username - password When using env vars, the names are: - SMTP_HOST - SMTP_PORT - SMTP_USERNAME - SMTP_PASSWORD - SMTP_FROM (defaults to SMTP_USERNAME) Optional: - SMTP_STARTTLS (default: "1") - SMTP_TIMEOUT_SECONDS (default: "10") - SMTP_DEBUG (default: "0") - set to 1 to print SMTP conversation to console """ # Load DB settings (best-effort). If the DB has a complete SMTP config, we will use # it even if environment variables are present. # # This fixes the common deployment situation where a .env/.flaskenv provides SMTP_* # values that unintentionally override admin-configured settings. db_defaults: dict[str, object] = {} try: # Local import to avoid import cycles and to keep this module lightweight. from flask import current_app from .extensions import db from .models import AppSettings # Only try if app context exists (send_email is called inside requests/CLI normally). # Accessing the LocalProxy will raise if there is no active app context. current_app._get_current_object() s = db.session.get(AppSettings, 1) if s: db_defaults = { "host": s.smtp_host, "port": s.smtp_port, "username": s.smtp_username, "password": s.smtp_password, "from_email": s.smtp_from, "starttls": s.smtp_starttls, "timeout": s.smtp_timeout_seconds, "debug": s.smtp_debug, } except Exception: # Best-effort; if DB isn't ready we fall back to env vars only. db_defaults = {} # DB-first selection rules: # - If *any* SMTP field is configured in DB, treat DB as authoritative (and do NOT # silently mix in env vars). This avoids confusion where partial DB config still # results in env vars being used. # - Env override is only possible with SMTP_OVERRIDE_DB=1. override_db = _truthy(os.environ.get("SMTP_OVERRIDE_DB")) db_config_present = bool( db_defaults.get("host") or db_defaults.get("username") or db_defaults.get("password") or db_defaults.get("from_email") or (db_defaults.get("port") is not None) ) if db_config_present and not override_db: config_source = "db" host = str(db_defaults.get("host") or "") or None port = int(db_defaults.get("port") or 587) username = str(db_defaults.get("username") or "") or None password = str(db_defaults.get("password") or "") or None from_email = (str(db_defaults.get("from_email")) if db_defaults.get("from_email") else None) or username starttls = bool(db_defaults.get("starttls") if db_defaults.get("starttls") is not None else True) timeout = float(db_defaults.get("timeout") or 10) debug = bool(db_defaults.get("debug") or False) else: config_source = "env" host = os.environ.get("SMTP_HOST") port = int(os.environ.get("SMTP_PORT") or 587) username = os.environ.get("SMTP_USERNAME") password = os.environ.get("SMTP_PASSWORD") from_email = os.environ.get("SMTP_FROM") or username starttls = _truthy(os.environ.get("SMTP_STARTTLS")) if os.environ.get("SMTP_STARTTLS") is not None else True timeout = float(os.environ.get("SMTP_TIMEOUT_SECONDS") or 10) debug = _truthy(os.environ.get("SMTP_DEBUG")) if os.environ.get("SMTP_DEBUG") is not None else False missing = [] if not host: missing.append("host") if not username: missing.append("username") if not password: missing.append("password") if not from_email: missing.append("from_email") if missing: raise RuntimeError( f"Missing SMTP configuration ({config_source}): " + ", ".join(missing) + ". Configure it in Admin → Settings (or set SMTP_* environment variables)." ) msg = EmailMessage() msg["From"] = from_email msg["To"] = to_email msg["Subject"] = subject # Helps when SMTP providers force the authenticated mailbox as envelope sender, # but still allow replies to go to the desired address. if from_email and username and from_email != username: msg["Reply-To"] = from_email msg.set_content(body_text) with smtplib.SMTP(host, port, timeout=timeout) as smtp: if debug: smtp.set_debuglevel(1) print(f"[email_utils] Using SMTP config from: {config_source}") smtp.ehlo() if starttls: smtp.starttls() smtp.ehlo() smtp.login(username, password) # Pass explicit envelope-from to avoid falling back to the authenticated user. # Note: some SMTP providers will still override this for anti-spoofing. smtp.send_message(msg, from_addr=from_email, to_addrs=[to_email])