"""Outbound email for onboarding (access requests + approval notices). Gmail SMTP + App Password by default. Deliberately fail-soft: if SMTP isn't configured or a send errors, we log and return False so the caller can fall back to in-app notifications — email is never load-bearing for the app's core flow. Fail-soft must not mean invisible, though: every drop also leaves an in-app notice in the admins' inbox (see `_alert_admins`). An instance whose SMTP was never filled in silently swallows verification links, password resets and access-request notices, and the operator's first clue used to be a person telling them nothing arrived. """ import logging import smtplib import ssl from email.message import EmailMessage from email.utils import formatdate from sqlalchemy import select from sqlalchemy.orm import Session from app import sysconfig from app.config import settings from app.db import SessionLocal from app.models import User from app.notifications import create_notification from app.ratelimit import RateLimiter log = logging.getLogger("siftlode.email") # One admin notice an hour per reason. A relay that is down fails every send, and the inbox must # stay readable; the log keeps the full record either way. _alert_throttle = RateLimiter(max_events=1, window_seconds=3600) def _alert_admins(reason: str, subject: str, recipients: list[str]) -> None: """Record an in-app notice that a message did NOT go out. Best effort in every direction: it opens its own session (callers are often background tasks), swallows its own errors, and never reports failure to the caller — this is the fallback path, it must not become a new way to break a send.""" if not _alert_throttle.allow(reason): return try: with SessionLocal() as db: admins = ( db.execute( select(User.id).where( User.role == "admin", User.is_active.is_(True), User.is_suspended.is_(False), ) ) .scalars() .all() ) if not admins: # Nothing was recorded, so don't let this cost the next hour of alerts on the very # path whose job is to stop drops going unseen. Same on the failure branch below. _alert_throttle.reset() return for uid in admins: create_notification( db, user_id=uid, type="email_undelivered", # English fallback; the inbox renders a translated version from `data`. title="Email could not be sent", body=subject, data={"reason": reason, "subject": subject, "recipients": recipients}, commit=False, ) db.commit() except Exception: _alert_throttle.reset() log.exception("Could not record the undelivered-email notice (%r)", subject) def _read_smtp(db: Session) -> dict: return { "host": sysconfig.get_str(db, "smtp_host"), "port": sysconfig.get_int(db, "smtp_port"), "user": sysconfig.get_str(db, "smtp_user"), "password": sysconfig.get_str(db, "smtp_password"), "from": sysconfig.get_str(db, "smtp_from"), } def _smtp(db: Session | None = None) -> dict: """Effective SMTP config (admin DB override over env defaults). Opens its own short session when the caller has none, because email is usually sent from a BackgroundTask outside a request's db. A caller that IS inside a request should pass its session: reading the password decrypts a Fernet value, and /api/me polls this for every admin tab.""" if db is not None: return _read_smtp(db) with SessionLocal() as own: return _read_smtp(own) def email_enabled(db: Session | None = None) -> bool: c = _smtp(db) return bool(c["host"] and c["user"] and c["password"]) def _admin_contact() -> str | None: return next(iter(sorted(settings.admin_email_set)), None) def _send(to: list[str], subject: str, body: str, reply_to: str | None = None) -> bool: recipients = [e for e in to if e] if not recipients: return False c = _smtp() if not (c["host"] and c["user"] and c["password"]): log.info("SMTP not configured; skipping email %r to %s", subject, recipients) _alert_admins("not_configured", subject, recipients) return False msg = EmailMessage() msg["Subject"] = subject msg["From"] = c["from"] or c["user"] msg["To"] = ", ".join(recipients) # A real Date and a Reply-To (so it's a conversation, not a no-reply blast) both # nudge spam filters the right way; reputation still does most of the work. msg["Date"] = formatdate(localtime=True) if reply_to: msg["Reply-To"] = reply_to msg.set_content(body) try: with smtplib.SMTP(c["host"], c["port"], timeout=20) as s: s.starttls(context=ssl.create_default_context()) s.login(c["user"], c["password"]) s.send_message(msg) log.info("Sent email %r to %s", subject, recipients) return True except Exception: log.exception("SMTP send failed (%r to %s)", subject, recipients) _alert_admins("send_failed", subject, recipients) return False def send_access_approved(email: str) -> bool: admin = _admin_contact() body = ( "Hi,\n\n" "Your request to use Siftlode has been approved — welcome aboard.\n\n" "Open Siftlode and sign in, and your YouTube subscriptions feed will be ready:\n\n" f"{settings.app_base}\n\n" "If you weren't expecting this, you can ignore the message.\n\n" "— Siftlode" + (f"\n\nQuestions? Just reply to this email ({admin})." if admin else "") ) return _send([email], "You're in — your Siftlode access is approved", body, reply_to=admin) def send_access_denied(email: str) -> bool: admin = _admin_contact() body = ( "Hi,\n\n" "Thanks for your interest in Siftlode. After review, your access request wasn't approved " "this time.\n\n" "If you think this was a mistake, you're welcome to reply and let us know.\n\n" "— Siftlode" + (f"\n\nQuestions? Just reply to this email ({admin})." if admin else "") ) return _send([email], "About your Siftlode access request", body, reply_to=admin) def send_role_changed(to: str, role: str, operator: str | None) -> bool: line = ( "You've been granted admin access on Siftlode — you can now manage settings, the " "scheduler and users." if role == "admin" else "Your admin access on Siftlode has been removed — you're now a regular user." ) contact = f"\n\nQuestions? Contact the operator at {operator}." if operator else "" body = "Hi,\n\n" + line + contact + "\n\n— Siftlode" return _send([to], "Your Siftlode role has changed", body, reply_to=operator) def send_account_unsuspended(to: str, operator: str | None) -> bool: contact = f"\n\nQuestions? Contact the operator at {operator}." if operator else "" body = ( "Hi,\n\n" "Your Siftlode account has been reinstated — the suspension was lifted and you can sign " "in again." + contact + "\n\n— Siftlode" ) return _send([to], "Your Siftlode account has been reinstated", body, reply_to=operator) def send_account_deleted(to: str, operator: str | None) -> bool: contact = ( f"\n\nIf you didn't request this, contact the Siftlode operator at {operator}." if operator else "" ) body = ( "Hi,\n\n" "Your Siftlode account and all associated data — subscriptions, tags, watch history, " "playlists and settings — have been permanently deleted. Nothing is retained.\n\n" "Thanks for having used Siftlode." + contact + "\n\n— Siftlode" ) return _send([to], "Your Siftlode account has been deleted", body, reply_to=operator) def send_account_suspended(to: str, operator: str | None) -> bool: contact = ( f"If you think this is a mistake or have questions, contact the Siftlode operator " f"at {operator}.\n\n" if operator else "If you think this is a mistake, please contact the Siftlode operator.\n\n" ) body = ( "Hi,\n\n" "A sign-in to your Siftlode account was just attempted, but the account is currently " "suspended, so access was denied.\n\n" + contact + "— Siftlode" ) return _send([to], "Your Siftlode account is suspended", body, reply_to=operator) def send_admin_new_request(admins: list[str], requester: str) -> bool: # The link deep-links straight to the admin Access-requests page (the SPA reads ?admin=…). The # requester address is spelled out (mail clients auto-link both the URL and the address), and the # reply note is accurate: _send sets Reply-To to the requester, so a reply reaches THEM — not the # sending mailbox. (The old copy said "Settings → Account", which was the wrong menu.) approve_url = f"{settings.app_base}/?admin=access-requests" body = ( f"{requester} just requested access to Siftlode.\n\n" f"Approve or deny it here:\n{approve_url}\n" f"(or in Siftlode: open Users → Access requests)\n\n" f"Requester's email: {requester}\n" f"Replying to this message goes straight to the requester (it's the Reply-To address).\n" ) return _send(admins, f"Siftlode access request from {requester}", body, reply_to=requester) def send_verify_email(to: str, link: str) -> bool: body = ( "Hi,\n\n" "Confirm your email to finish creating your Siftlode account:\n\n" f"{link}\n\n" "If you didn't sign up, you can ignore this message.\n\n" "— Siftlode" ) return _send([to], "Confirm your Siftlode email", body) def send_password_reset(to: str, link: str) -> bool: body = ( "Hi,\n\n" "Use this link to set a new Siftlode password (it expires in an hour):\n\n" f"{link}\n\n" "If you didn't request this, you can ignore this message — your password is unchanged.\n\n" "— Siftlode" ) return _send([to], "Reset your Siftlode password", body) def send_test(to: str) -> bool: body = ( "This is a test email from Siftlode.\n\n" "If you're reading this, your SMTP settings work.\n\n" "— Siftlode" ) return _send([to], "Siftlode SMTP test", body)