feat(notifications): JSON-safe webhook template, SMTP timeout and implicit TLS, test-send buttons (#5064)

#5064 "Notifications that arrive":
- render_webhook: alert.* is escaped to sit inside a JSON string (a quote
  or newline in a rule name no longer breaks the body); raw.* is the plain
  value for |tojson; a missing value reads "no data"; the template runs
  sandboxed and the body is checked to be JSON before it is sent.
- SMTP: 15s timeout; port 465 connects with implicit TLS, otherwise the
  box means STARTTLS; a From address setting (falls back to the username).
- A webhook error no longer carries the URL, which is a secret.
- Settings → Notifications: "Send a test email" / "Send a test webhook"
  send a sample alert with the form's values (unsaved) and show the result
  and the body that went out; template help names alert.* and raw.*.
- Notification audits go through audit_request; the test send is audited
  as settings.notification.tested, with the channel and outcome only.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

steward/settings/routes.py

@@ -5,7 +5,7 @@ from pathlib import Path
 from quart import Blueprint, current_app, render_template, request, redirect, url_for, session
 
 from steward.auth.middleware import require_role
-from steward.core.audit import log_audit
+from steward.core.audit import audit_request, log_audit
 from steward.models.users import UserRole
 from steward.core.settings import (
     get_all_settings, set_setting, delete_setting,
@@ -229,17 +229,86 @@ async def save_notifications():
             if form.get("smtp.password"):
                 await set_setting(db, "smtp.password", form.get("smtp.password"))
             await set_setting(db, "smtp.recipients", recipients)
+            await set_setting(db, "smtp.from", form.get("smtp.from", "").strip())
             if "webhook.remove" in form:
                 await set_setting(db, "webhook.url", "")
             elif form.get("webhook.url", "").strip():
                 await set_setting(db, "webhook.url", form.get("webhook.url").strip())
             await set_setting(db, "webhook.template", form.get("webhook.template", ""))
     await _reload_app_config()
-    await log_audit(current_app, session.get("user_id"), session.get("username", ""),
-                    "settings.saved", detail={"section": "notifications"})
+    await audit_request("settings.saved", detail={"section": "notifications"})
     return redirect(url_for("settings.notifications"))
 
 
+def _sample_alert() -> dict:
+    """What a test notification carries: a made-up firing alert whose name
+    has quotes in it, so a template that cannot take them shows it."""
+    from datetime import datetime, timezone
+    return {"rule_name": 'Test alert "from" Steward', "state": "TEST", "metric": "cpu_pct",
+            "value": 97.5, "threshold": 90.0, "resource": "example-host",
+            "source_module": "host_agent", "note": "sent from Settings → Notifications",
+            "timestamp": datetime.now(timezone.utc).isoformat()}
+
+
+async def _notify_test_result(channel: str, ok: bool, result: str, payload: str | None = None):
+    await audit_request("settings.notification.tested", detail={"channel": channel, "sent": ok})
+    return await render_template("settings/_notify_test.html", ok=ok, result=result,
+                                 payload=payload)
+
+
+@settings_bp.post("/notifications/test-webhook")
+@require_role(UserRole.admin)
+async def test_webhook():
+    """Render the webhook template (as in the form, unsaved) for a sample
+    alert and send it to the URL in the form, or the saved one."""
+    from steward.core.notifications import _send_webhook, render_webhook
+    form = await request.form
+    async with current_app.db_sessionmaker() as db:
+        saved = to_webhook_cfg(await get_all_settings(db))
+    url = (form.get("webhook.url") or "").strip() or saved["url"]
+    template = form.get("webhook.template") or saved["template"]
+    alert = _sample_alert()
+    try:
+        payload = render_webhook(template, alert)
+    except ValueError as exc:
+        return await _notify_test_result("webhook", False, str(exc))
+    if not url:
+        return await _notify_test_result(
+            "webhook", False, "No webhook URL is saved or entered. This is the body it would get.",
+            payload)
+    sent = await _send_webhook({"url": url, "template": template}, alert)
+    return await _notify_test_result(
+        "webhook", sent["sent"], "Delivered." if sent["sent"] else sent["error"], payload)
+
+
+@settings_bp.post("/notifications/test-email")
+@require_role(UserRole.admin)
+async def test_email():
+    """Send a sample alert email using the SMTP settings in the form (unsaved);
+    a blank password uses the saved one."""
+    from steward.core.notifications import _send_email, email_body, smtp_mode
+    form = await request.form
+    async with current_app.db_sessionmaker() as db:
+        saved = to_smtp_cfg(await get_all_settings(db))
+    try:
+        port = int(form.get("smtp.port") or 587)
+    except ValueError:
+        return await _notify_test_result("email", False, "The port must be a number.")
+    cfg = {"host": (form.get("smtp.host") or "").strip(), "port": port,
+           "tls": "smtp.tls" in form, "username": form.get("smtp.username", ""),
+           "password": form.get("smtp.password") or saved["password"],
+           "from": (form.get("smtp.from") or "").strip(),
+           "recipients": [r.strip() for r in form.get("smtp.recipients", "").split(",") if r.strip()]}
+    if not cfg["host"] or not cfg["recipients"]:
+        return await _notify_test_result("email", False, "Enter an SMTP host and at least one recipient.")
+    alert = _sample_alert()
+    sent = await _send_email(cfg, alert)
+    how = {"ssl": "implicit TLS", "starttls": "STARTTLS", "plain": "no TLS"}[smtp_mode(port, cfg["tls"])]
+    result = (f"Sent to {', '.join(cfg['recipients'])} via {cfg['host']}:{port} ({how})."
+              if sent["sent"] else f"{cfg['host']}:{port} ({how}): {sent['error']}")
+    return await _notify_test_result("email", sent["sent"], result, email_body(alert))
+
+
 # ── Ansible Sources ───────────────────────────────────────────────────────────
 
 async def _get_ansible_sources() -> list[dict]: