"""Notifications: admin-configurable email + an outbox. Email notifications are OFF by default and controlled from the admin console (a toggle stored in `app_settings`). Even when enabled, mail is only sent if SMTP is configured. The SMTP PASSWORD is read from the `SMTP_PASSWORD` environment variable and is NEVER stored in the database or shown in the UI. Every notable event (e.g. a WP assignment) writes a `notifications` row — an in-app record — and, when email is on + SMTP is set, the row is delivered by email in a background task. Notification bodies carry customer CONTEXT — the WP number, its title, where the work happens — and a deep link, never customer document CONTENT (scope text, descriptions, comments, attachments). Decided 2026-08-20; the link is the summary of everything a body leaves out. """ import os import smtplib import socket import uuid import logging from email.message import EmailMessage from typing import Optional from sqlalchemy.orm import Session from . import models log = logging.getLogger("wpsuite.notify") SETTINGS_KEY = "notifications" DEFAULTS = { "email_enabled": False, # master toggle — OFF until SMTP is sorted "smtp_host": "", "smtp_port": 587, "smtp_use_tls": True, "smtp_username": "", "from_addr": "", "from_name": "Work Package Suite", "app_base_url": "", # e.g. https://wp.controls.dev — used to build email links # Feature flags (admin console). BIM/VDC is off until it's ready for the field: # with it off, the SOP creator hides the BIM section entirely and every SOP is # install-only, so no project can be put on the BIM path by accident. "bim_enabled": False, # Localization defaults for dates, times and numbers. Empty = use each # browser's own locale / timezone. A user's own preference wins over these. "default_locale": "", # BCP47, e.g. en-US "default_timezone": "", # IANA, e.g. America/Chicago } # Settings the app needs before anyone is signed in, or that carry no secrets and # are safe for any authenticated user to read (feature flags + localization # defaults + whether self-service password reset can work at all). PUBLIC_KEYS = ("bim_enabled", "default_locale", "default_timezone") def get_settings(db: Session) -> dict: row = db.get(models.AppSetting, SETTINGS_KEY) s = dict(DEFAULTS) if row and row.value: s.update({k: row.value[k] for k in row.value if k in DEFAULTS}) return s def save_settings(db: Session, patch: dict) -> dict: cur = get_settings(db) for k in DEFAULTS: if k in patch and patch[k] is not None: cur[k] = patch[k] row = db.get(models.AppSetting, SETTINGS_KEY) if row: row.value = cur else: db.add(models.AppSetting(key=SETTINGS_KEY, value=cur)) db.commit() return cur def public_settings(db: Session) -> dict: """Settings safe to return to the admin UI — no secrets.""" s = get_settings(db) s["smtp_password_set"] = bool(os.getenv("SMTP_PASSWORD")) return s def app_flags(db: Session) -> dict: """Feature flags for any signed-in user (no secrets, no SMTP detail). `password_reset_enabled` tells the login page whether a self-service reset can actually deliver mail — there's no point offering the link otherwise.""" s = get_settings(db) out = {k: s.get(k) for k in PUBLIC_KEYS} out["password_reset_enabled"] = bool(s.get("email_enabled")) and smtp_ready(s) return out def smtp_ready(s: dict) -> bool: return bool(s.get("smtp_host") and s.get("from_addr")) def send_email(s: dict, to_addr: str, subject: str, body: str) -> None: """Send one email via SMTP. Raises on any failure (caller records it).""" if not to_addr: raise ValueError("no recipient email") msg = EmailMessage() from_name = s.get("from_name") or "" msg["From"] = f"{from_name} <{s['from_addr']}>" if from_name else s["from_addr"] msg["To"] = to_addr msg["Subject"] = subject msg.set_content(body) host = s["smtp_host"] port = int(s.get("smtp_port") or 587) user = s.get("smtp_username") or "" pw = os.getenv("SMTP_PASSWORD", "") # local_hostname pins the EHLO name. Without it smtplib calls getfqdn() on # EVERY connect, and that reverse-DNS lookup stalls ~5s per send whenever DNS # is slow or unreachable - sends are sequential background tasks, so a batch # of notifications trickled out one per five seconds. gethostname() never # touches the network. Found 2026-08-20 when the office link dropped. with smtplib.SMTP(host, port, timeout=15, local_hostname=(socket.gethostname() or "wp-suite")) as srv: if s.get("smtp_use_tls", True): srv.starttls() if user: srv.login(user, pw) srv.send_message(msg) def send_now(db: Session, to_addr: str, subject: str, body: str) -> bool: """Send one email immediately, outside the outbox. Used for password resets — a reset link must never sit in a queue, and it must not be persisted in the notifications table where an admin could read it and take over the account. Returns True if it went out.""" s = get_settings(db) if not (s.get("email_enabled") and smtp_ready(s) and to_addr): return False try: send_email(s, to_addr, subject, body) return True except Exception as e: # noqa: BLE001 — never surface SMTP detail to the caller log.warning("password-reset email to %s failed: %s", to_addr, e) return False def enqueue(db: Session, *, user: "models.User", kind: str, subject: str, body: str, link: str = "", wp_id: Optional[str] = None, project_id: Optional[str] = None) -> "models.Notification": """Record a notification. Marked 'pending' only if email is enabled + SMTP ready + the recipient has an email; otherwise 'skipped' (still an in-app record). Does NOT commit — the caller commits with its own transaction. Returns the row.""" s = get_settings(db) deliverable = bool(s.get("email_enabled")) and smtp_ready(s) and bool(user.email) n = models.Notification( id="ntf_" + uuid.uuid4().hex[:12], user_id=user.id, email=user.email or "", kind=kind, wp_id=wp_id, project_id=project_id, subject=subject[:300], body=body, link=link[:500], status="pending" if deliverable else "skipped", ) db.add(n) return n def deliver(notif_id: str) -> None: """Background task: send one pending notification, on its own DB session.""" from .db import SessionLocal db = SessionLocal() try: n = db.get(models.Notification, notif_id) if not n or n.status != "pending": return s = get_settings(db) if not (s.get("email_enabled") and smtp_ready(s) and n.email): n.status = "skipped" db.commit() return try: send_email(s, n.email, n.subject, n.body) n.status = "sent" n.sent_at = models.utcnow() except Exception as e: # noqa: BLE001 — record any SMTP failure, don't crash the worker n.status = "failed" n.error = str(e)[:400] log.warning("notification %s failed to send: %s", notif_id, e) db.commit() finally: db.close()