Project-SDE-WP-Suite/server/notify.py

"""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 deliberately avoid customer IP: they carry a WP
number and a deep link, not the work-package contents.
"""
import os
import smtplib
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", "")
    with smtplib.SMTP(host, port, timeout=15) 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()