diff --git a/docs/core/alerting.md b/docs/core/alerting.md index 5ee6a14..dbfc94d 100644 --- a/docs/core/alerting.md +++ b/docs/core/alerting.md @@ -168,10 +168,11 @@ alert in the window, oldest first, in their own zone, with a subject like "[Steward] 3 alerts: 2 firing, 1 resolved". One alert on its own keeps its usual email. -Email waits in the database (`email_outbox`, `core/outbox.py`) until the mail -server accepts it, so nothing waiting is lost when Steward restarts: the -sender picks it up at boot. The sender runs every 15 seconds and settles each -email on the server's answer: +Every email Steward sends (alerts, the weekly report, invites, password +resets and the test buttons) waits in the database (`email_outbox`, +`core/outbox.py`) until the mail server accepts it, so nothing waiting is lost +when Steward restarts: the sender picks it up at boot. The sender runs every 15 +seconds and settles each email on the server's answer: - **Accepted:** sent. - **Refused** (an SMTP 5xx, such as "no such mailbox"): failed at once, with @@ -183,6 +184,13 @@ email on the server's answer: Once every email carrying an alert has settled, that alert's event records whether the email went, and to whom it didn't. +An email a person is waiting on is tried at once instead of on the next tick: +the weekly report's **Send now**, an invite, and the test buttons, so the page +can say what the server answered. A report or invite the server couldn't take +yet stays queued and is retried as above. A test (the Settings → Notifications +test email, a rule's **Send test**) is recorded but never retried: its one +answer is the result. + ### Webhook Configured at `/settings/` under Webhook. The body is a Jinja2 template rendered to JSON. Content-Type is always `application/json`. If the rendered template is not valid JSON, the delivery is logged as failed and no request is sent. diff --git a/steward/alerts/service.py b/steward/alerts/service.py index aecb2b7..a5b8b43 100644 --- a/steward/alerts/service.py +++ b/steward/alerts/service.py @@ -266,9 +266,9 @@ async def active_count(db) -> int: def notify_targets(smtp: dict, webhook: dict, accounts: int = 0) -> list[str]: """Where an alert's notifications go, in words, by the same tests - core.notifications.dispatch_notifications sends by: email when a server - is set and `accounts` (that get alerts) or Also send to addresses exist, - a webhook when its URL is. Empty: nobody (#5315, milestone 495).""" + alerts are sent by: email when a server is set and `accounts` (that get + alerts) or Also send to addresses exist (core.outbox.hold_alert), a + webhook when its URL is (core.notifications.dispatch_notifications). Empty: nobody (#5315, milestone 495).""" from urllib.parse import urlsplit out = [] recipients = smtp.get("recipients") or [] @@ -752,7 +752,8 @@ async def send_test(app, rule_id: str) -> dict: failed; lines say what happened to each channel. LookupError for no rule. """ from urllib.parse import urlsplit - from steward.core.mail import recipients + from steward.core import outbox + from steward.core.mail import alert_email, recipients from steward.core.notifications import dispatch_notifications smtp, webhook = app.config.get("SMTP", {}), app.config.get("WEBHOOK", {}) async with app.db_sessionmaker() as db: @@ -761,12 +762,14 @@ async def send_test(app, rule_id: str) -> dict: raise LookupError(rule_id) alert = await test_alert(db, rule) to = await recipients(db, "alerts", smtp) if smtp.get("host") else [] - results = await dispatch_notifications(smtp, webhook, alert, to) + results = await dispatch_notifications(webhook, alert) lines = [] - if email := results.get("email"): - who = ", ".join(r.address for r in to) - lines.append(f"Email sent to {who}." if email["sent"] - else f"Email to {who} failed: {email['error']}") + if to: + # Queued and recorded like any email, tried once, not retried. + ok, said = outbox.said(await outbox.send_each( + app, "test", to, lambda r: alert_email(smtp, alert, r), retry=False)) + results["email"] = {"sent": ok, "error": None if ok else said} + lines.append(f"Email: {said}") if hook := results.get("webhook"): host = urlsplit(webhook["url"]).hostname or webhook["url"] lines.append(f"Webhook to {host} delivered." if hook["sent"] diff --git a/steward/auth/routes.py b/steward/auth/routes.py index 8f2fd4a..066f692 100644 --- a/steward/auth/routes.py +++ b/steward/auth/routes.py @@ -436,7 +436,7 @@ async def forgot_post(): accounts; every request from an address counts toward its limit.""" from quart import current_app from steward.auth.throttle import reset_throttle - from steward.core import mail + from steward.core import mail, outbox from steward.core.audit import log_audit from steward.core.formatting import zone from steward.core.password_resets import account_for, issue, link @@ -454,16 +454,15 @@ async def forgot_post(): user = await account_for(db, name) if user is not None: reset, key = await issue(db, user) - to, username, user_id, tz = user.email, user.username, user.id, user.time_zone + username, user_id = user.username, user.id + to = mail.Recipient(user.email, zone(user.time_zone), False, username) + subject, text, html = mail.reset_email(username, reset.expires_at, + link(public_base_url(request), key), to.tz) + # Sent by the outbox, with the link's own transaction; the page + # says the same either way, and the queue records how it went. + outbox.enqueue(db, kind="reset", to=to, subject=subject, text=text, html=html) if user is None: return await _forgot_page(sent=True) - cfg = current_app.config.get("SMTP", {}) - subject, text, html = mail.reset_email(username, reset.expires_at, - link(public_base_url(request), key), zone(tz)) - try: - await mail.send(cfg, [mail.message(cfg, to, subject, text, html)]) - except Exception as exc: # the page still says the same; the log says why - current_app.logger.warning("Password reset email to %s failed: %s", to, exc) await log_audit(current_app, user_id, username, "user.password_reset_requested", "user", user_id, detail={"address": ip}) return await _forgot_page(sent=True) diff --git a/steward/core/alerts.py b/steward/core/alerts.py index e8a1bf8..f656e9f 100644 --- a/steward/core/alerts.py +++ b/steward/core/alerts.py @@ -522,7 +522,7 @@ async def _dispatch_notification( await outbox.hold_alert(app, alert_data, event_id) except Exception: logger.exception("Alert %s: its email could not be queued", event_id) - results = await dispatch_notifications({}, app.config.get("WEBHOOK", {}), alert_data) + results = await dispatch_notifications(app.config.get("WEBHOOK", {}), alert_data) webhook_r = results.get("webhook") if not webhook_r: diff --git a/steward/core/mail.py b/steward/core/mail.py index 7c65f7c..720dbac 100644 --- a/steward/core/mail.py +++ b/steward/core/mail.py @@ -163,13 +163,6 @@ async def audience(db) -> dict: return out -async def send(cfg: dict, messages: list[EmailMessage]) -> None: - """Send every message over one SMTP connection, off the event loop.""" - import asyncio - from steward.core.notifications import _smtp_send_sync - await asyncio.get_running_loop().run_in_executor(None, _smtp_send_sync, cfg, messages) - - def _render(name: str, **ctx) -> tuple[str, str]: ctx = {"c": palette(), **ctx} return (_env.get_template(f"{name}.txt").render(ctx), diff --git a/steward/core/notifications.py b/steward/core/notifications.py index 814fcda..62e4056 100644 --- a/steward/core/notifications.py +++ b/steward/core/notifications.py @@ -4,7 +4,6 @@ import logging import smtplib import ssl import types -from email.message import EmailMessage import httpx from jinja2.sandbox import SandboxedEnvironment @@ -12,20 +11,11 @@ from jinja2.sandbox import SandboxedEnvironment logger = logging.getLogger(__name__) -async def dispatch_notifications( - smtp_cfg: dict, - webhook_cfg: dict, - alert_data: dict, - recipients: list = (), -) -> dict: - """Dispatch all configured notification channels: email to each of - `recipients` (mail.Recipient, from mail.recipients) when a server is set. - Returns results dict.""" +async def dispatch_notifications(webhook_cfg: dict, alert_data: dict) -> dict: + """Post an alert's webhook when one is set; {"webhook": result} or {}. + Email goes through the outbox instead (core.outbox, #5506).""" results = {} - if smtp_cfg.get("host") and recipients: - results["email"] = await _send_email(smtp_cfg, alert_data, recipients) - if webhook_cfg.get("url"): results["webhook"] = await _send_webhook(webhook_cfg, alert_data) @@ -43,21 +33,10 @@ def smtp_mode(port, tls) -> str: return "starttls" if tls else "plain" -async def _send_email(cfg: dict, alert: dict, recipients: list) -> dict: - """Each recipient their own copy, in their own zone, over one connection.""" - from steward.core.mail import alert_email, message, send - try: - messages = [message(cfg, to.address, *alert_email(cfg, alert, to)) for to in recipients] - await send(cfg, messages) - return {"sent": True, "error": None} - except Exception as exc: - logger.error(f"Email notification failed: {exc}") - return {"sent": False, "error": str(exc)} - - def smtp_open(cfg: dict) -> smtplib.SMTP: """A connection to the mail server, TLS and login done, every step - bounded by SMTP_TIMEOUT_S. The caller closes it (use it in `with`).""" + bounded by SMTP_TIMEOUT_S. The caller closes it (use it in `with`). + Only core.outbox.deliver_sync opens one: every email is a queued row.""" port = int(cfg.get("port") or 587) mode = smtp_mode(port, cfg.get("tls", True)) ctx = ssl.create_default_context() @@ -76,12 +55,6 @@ def smtp_open(cfg: dict) -> smtplib.SMTP: return smtp -def _smtp_send_sync(cfg: dict, messages: list[EmailMessage]) -> None: - with smtp_open(cfg) as smtp: - for msg in messages: - smtp.send_message(msg) - - def _json_str(value) -> str: """A value as it may sit between the quotes of a JSON string.""" return json.dumps("" if value is None else str(value))[1:-1] diff --git a/steward/core/outbox.py b/steward/core/outbox.py index 4f460d3..3d2a427 100644 --- a/steward/core/outbox.py +++ b/steward/core/outbox.py @@ -33,7 +33,7 @@ import logging import smtplib from datetime import datetime, timedelta, timezone from email.message import EmailMessage -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Callable from sqlalchemy import insert, or_, select, text, update @@ -104,18 +104,22 @@ def reason(exc: BaseException) -> str: def enqueue(db, *, kind: str, to: "Recipient | str", subject: str | None = None, text: str | None = None, html: str | None = None, - send_after: datetime | None = None, retry: bool = True, + send_after: datetime | None = None, retry: bool = True, claimed: bool = False, now: datetime | None = None) -> EmailOutbox: """Add one email to the queue in `db`'s transaction (the caller commits). `send_after` holds it until then (an alert burst, #5507); otherwise it is - due at once. `to` is a mail.Recipient, or a bare address read in UTC.""" + due at once. `claimed` marks it as being sent by the caller, who will + `attempt` it right after committing, so the sender doesn't take it too; + a claim the caller never settles goes back to the queue (STALE_CLAIM_S). + `to` is a mail.Recipient, or a bare address read in UTC.""" now = now or datetime.now(timezone.utc) address, tz, show_utc = (to, "UTC", False) if isinstance(to, str) else \ (to.address, getattr(to.tz, "key", None) or "UTC", to.show_utc) row = EmailOutbox(kind=kind, recipient=address, tz=tz, show_utc=show_utc, subject=subject, text=text, html=html, retry=retry, created_at=now, - state="held" if send_after else "pending", send_after=send_after, - next_attempt_at=send_after or now) + state="held" if send_after else "sending" if claimed else "pending", + send_after=send_after, next_attempt_at=send_after or now, + claimed_at=now if claimed else None) db.add(row) return row @@ -219,14 +223,18 @@ async def write_held(app: "Quart", db, row: EmailOutbox) -> None: row.subject, row.text, row.html = alert_words(app.config.get("SMTP", {}), alerts, to) -async def attempt(app: "Quart", ids: list[str], now: datetime | None = None) -> dict[str, str]: +async def attempt(app: "Quart", ids: list[str], now: datetime | None = None, + cfg: dict | None = None) -> dict[str, EmailOutbox]: """Try to send the claimed rows `ids` once, and settle each. The words are written and kept first, so a retry sends the same email; no session - is held while the mail server is talked to. Returns {id: state after}.""" + is held while the mail server is talked to. `cfg` replaces the saved mail + settings for this one try (a test of the form's unsaved values; such a + row has retry=False, so nothing later needs them). Returns {id: row} + as settled.""" from steward.core.mail import message if not ids: return {} - cfg = app.config.get("SMTP", {}) + cfg = cfg if cfg is not None else app.config.get("SMTP", {}) outcomes: dict[str, BaseException | None] = {} sendable: list[tuple[str, EmailMessage]] = [] async with app.db_sessionmaker() as db: @@ -257,7 +265,7 @@ async def attempt(app: "Quart", ids: list[str], now: datetime | None = None) -> if row.state == "failed": logger.warning("email %s to %s failed: %s", row.kind, row.recipient, row.last_error) await after_settle(db, rows) - return {row.id: row.state for row in rows} + return {row.id: row for row in rows} async def after_settle(db, rows: list[EmailOutbox]) -> None: @@ -315,6 +323,36 @@ async def send_due(app: "Quart") -> None: now = datetime.now(timezone.utc) ids = await claim_due(app, now) if ids: - states = await attempt(app, ids, now) - logger.info("email outbox: %d tried, %d sent", len(states), - sum(1 for s in states.values() if s == "sent")) + rows = await attempt(app, ids, now) + logger.info("email outbox: %d tried, %d sent", len(rows), + sum(1 for r in rows.values() if r.state == "sent")) + + +async def send_each(app: "Quart", kind: str, to: list, words: Callable[..., tuple], + *, retry: bool = True, cfg: dict | None = None) -> list[EmailOutbox]: + """Queue one email per reader in `to` (mail.Recipient or address), + written by `words(reader)` -> (subject, text, html), and try them at + once, for a page that answers with what happened (see `said`). Returns + the rows as settled: a temporary failure is left pending for the sender + to retry, unless retry=False (a test). `cfg` as for `attempt`.""" + async with app.db_sessionmaker() as db: + async with db.begin(): + rows = [enqueue(db, kind=kind, to=r, subject=s, text=t, html=h, retry=retry, claimed=True) + for r in to for s, t, h in [words(r)]] + await db.flush() + ids = [r.id for r in rows] + done = await attempt(app, ids, cfg=cfg) + return [done[i] for i in ids if i in done] + + +def said(rows: list[EmailOutbox]) -> tuple[bool, str]: + """(all sent, what happened) for rows just tried: who it went to, who + refused and why, and who Steward is still trying.""" + sent = [r.recipient for r in rows if r.state == "sent"] + failed = [f"{r.recipient}: {r.last_error}" for r in rows if r.state == "failed"] + later = [f"{r.recipient} ({r.last_error})" for r in rows if r.state == "pending"] + parts = ([f"Sent to {', '.join(sent)}."] if sent else []) + \ + ([f"Not sent to {'; '.join(failed)}."] if failed else []) + \ + ([f"Not sent yet to {'; '.join(later)}; Steward will try again for up to an hour."] + if later else []) + return bool(rows) and not failed and not later, " ".join(parts) diff --git a/steward/core/reports.py b/steward/core/reports.py index f3a88f4..21634fb 100644 --- a/steward/core/reports.py +++ b/steward/core/reports.py @@ -11,7 +11,7 @@ from sqlalchemy import and_, case, func, select from steward.alerts import service as alert_svc from steward.core.formatting import zone -from steward.core.mail import message, recipients, report_email, send +from steward.core.mail import recipients, report_email from steward.core.settings import get_setting from steward.models.alerts import AlertEvent from steward.models.metrics import PluginMetric @@ -134,13 +134,13 @@ async def send_report(app) -> tuple[bool, str]: logger.exception("Failed to build report data") return False, f"Data collection failed: {exc}" - try: - await send(smtp_cfg, [message(smtp_cfg, r.address, *report_email(smtp_cfg, data, r)) for r in to]) - except Exception as exc: - logger.error("Failed to send weekly report: %s", exc) - return False, f"Send failed: {exc}" - logger.info("Weekly report sent to %d address(es)", len(to)) - return True, f"Report sent to {', '.join(r.address for r in to)}." + from steward.core import outbox + # Queued, then tried at once so the button can say what happened; one + # the server didn't take yet stays queued and is retried for an hour. + ok, said = outbox.said(await outbox.send_each( + app, "report", to, lambda r: report_email(smtp_cfg, data, r))) + logger.info("Weekly report: %s", said) + return ok, said def report_due(now: datetime, day: int, hour: int, tz) -> bool: diff --git a/steward/settings/routes.py b/steward/settings/routes.py index 19c2b33..3247dce 100644 --- a/steward/settings/routes.py +++ b/steward/settings/routes.py @@ -319,7 +319,8 @@ async def test_email(): the zone chosen for them. A blank password uses the saved one.""" from steward.core.formatting import zone from steward.core.mail import Recipient, alert_email, is_real_address, pick_recipients - from steward.core.notifications import _send_email, smtp_mode + from steward.core import outbox + from steward.core.notifications import smtp_mode from steward.models.users import User form = await request.form async with current_app.db_sessionmaker() as db: @@ -345,11 +346,12 @@ async def test_email(): "email", False, "Your account has no email address mail can reach, and no Also send to " "addresses are entered. Set your address on the Account page.") alert = _sample_alert() - sent = await _send_email(cfg, alert, to) + # Queued and recorded like any email, tried once with the form's values. + ok, said = outbox.said(await outbox.send_each( + current_app, "test", to, lambda r: alert_email(cfg, alert, r), retry=False, cfg=cfg)) how = {"ssl": "implicit TLS", "starttls": "STARTTLS", "plain": "no TLS"}[smtp_mode(port, cfg["tls"])] - result = (f"Sent to {', '.join(r.address for r in to)} 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, alert_email(cfg, alert, to[0])[1]) + result = f"{said[:-1]} via {cfg['host']}:{port} ({how})." if ok else f"{cfg['host']}:{port} ({how}): {said}" + return await _notify_test_result("email", ok, result, alert_email(cfg, alert, to[0])[1]) # ── Ansible Sources ─────────────────────────────────────────────────────────── diff --git a/steward/settings/users.py b/steward/settings/users.py index d674475..e536b15 100644 --- a/steward/settings/users.py +++ b/steward/settings/users.py @@ -185,12 +185,15 @@ async def _deliver(invite: UserInvite, key: str) -> str | None: url = invites.link(public_base_url(request), key) subject, text, html = mail.invite_email(session.get("username", "An admin"), invite.role.value, invite.expires_at, url, zone(session.get("tz"))) - try: - await mail.send(cfg, [mail.message(cfg, invite.email, subject, text, html)]) - except Exception as exc: - current_app.logger.warning("Invite email to %s failed: %s", invite.email, exc) - return f"Sending to {invite.email} failed ({exc})." - return None + from steward.core import outbox + (row,) = await outbox.send_each(current_app, "invite", [invite.email], + lambda _: (subject, text, html)) + if row.state == "sent": + return None + if row.state == "pending": + return (f"Sending to {invite.email} didn't go through yet ({row.last_error}); " + f"Steward will keep trying for up to an hour.") + return f"Sending to {invite.email} failed ({row.last_error})." async def _link_page(invite: UserInvite, key: str, why: str): diff --git a/tests/core/test_mail.py b/tests/core/test_mail.py index 7c0648a..e3a9014 100644 --- a/tests/core/test_mail.py +++ b/tests/core/test_mail.py @@ -6,7 +6,6 @@ from datetime import datetime, timezone from zoneinfo import ZoneInfo from steward.core import mail -from steward.core import notifications as n from steward.core import reports ALERT = {"rule_name": 'Disk "root" on web01', "state": "FIRING", "metric": "disk_used_pct", @@ -46,9 +45,8 @@ def test_every_colour_is_a_theme_value(): async def _sent(monkeypatch, cfg, alert, to=(EXTRA_NY,)): - out = [] - monkeypatch.setattr(n, "_smtp_send_sync", lambda c, msgs: out.extend(msgs)) - assert (await n._send_email(cfg, alert, list(to)))["sent"] + """The message each of `to` is handed for `alert`, as the outbox writes it.""" + out = [mail.message(cfg, r.address, *mail.alert_email(cfg, alert, r)) for r in to] return out[0] if len(to) == 1 else out @@ -140,12 +138,21 @@ async def test_the_report_is_sent_from_the_configured_address(monkeypatch): assert kind == "report" return [PERSON_NY, EXTRA_NY] + from types import SimpleNamespace + from steward.core import outbox out = [] + + async def _send_each(app, kind, to, words, **kw): + # Queued then tried at once (#5508); here every one is accepted. + assert kind == "report" and kw.get("retry", True) + out.extend(mail.message(app.config["SMTP"], r.address, *words(r)) for r in to) + return [SimpleNamespace(recipient=r.address, state="sent", last_error=None) for r in to] + monkeypatch.setattr(reports, "build_report_data", _data) monkeypatch.setattr(reports, "recipients", _recipients) - monkeypatch.setattr(n, "_smtp_send_sync", lambda c, msgs: out.extend(msgs)) + monkeypatch.setattr(outbox, "send_each", _send_each) ok, said = await reports.send_report(_App()) - assert ok and said == "Report sent to ann@example.com, ops@example.com." + assert ok and said == "Sent to ann@example.com, ops@example.com." assert [(m["From"], m["To"]) for m in out] == [("alerts@example.com", "ann@example.com"), ("alerts@example.com", "ops@example.com")] assert all(m.get_content_type() == "multipart/alternative" for m in out) diff --git a/tests/core/test_notifications.py b/tests/core/test_notifications.py index f969165..2d388bd 100644 --- a/tests/core/test_notifications.py +++ b/tests/core/test_notifications.py @@ -49,7 +49,7 @@ def test_a_bad_template_raises_valueerror_saying_why(tpl, says): n.render_webhook(tpl, ALERT) -# ── smtp_mode / _smtp_send_sync ─────────────────────────────────────────────── +# ── smtp_mode / smtp_open (via outbox.deliver_sync) ─────────────────────────────────────────────── @pytest.mark.parametrize("port, tls, mode", [ (465, True, "ssl"), (465, False, "ssl"), ("465", False, "ssl"), @@ -98,27 +98,30 @@ class SMTP_SSL(_FakeSMTP): (25, False, [("connect", "SMTP", "mx", 25, n.SMTP_TIMEOUT_S)]), ]) def test_smtp_send_connects_by_mode_with_a_timeout(monkeypatch, port, tls, expected): + from email.message import EmailMessage + from steward.core import outbox monkeypatch.setattr(n.smtplib, "SMTP", SMTP) monkeypatch.setattr(n.smtplib, "SMTP_SSL", SMTP_SSL) _FakeSMTP.made = [] - msgs = [n.EmailMessage(), n.EmailMessage()] + msgs = [EmailMessage(), EmailMessage()] msgs[0]["From"], msgs[1]["From"] = "steward@example.com", "other@example.com" - n._smtp_send_sync({"host": "mx", "port": port, "tls": tls, "username": "u", "password": "p"}, msgs) + cfg = {"host": "mx", "port": port, "tls": tls, "username": "u", "password": "p"} + assert outbox.deliver_sync(cfg, msgs) == [None, None] (smtp,) = _FakeSMTP.made # every message over one connection assert smtp.calls == expected + [("login", "u", "p"), ("send", "steward@example.com"), ("send", "other@example.com")] -async def test_email_from_falls_back_and_a_line_break_in_the_name_still_sends(monkeypatch): +def test_email_from_falls_back_and_a_line_break_in_the_name_still_sends(): + from steward.core.mail import alert_email, message sent = [] - monkeypatch.setattr(n, "_smtp_send_sync", - lambda cfg, msgs: sent.extend((m["From"], m["Subject"]) for m in msgs)) - to = [Recipient("a@example.com", timezone.utc, False)] + to = Recipient("a@example.com", timezone.utc, False) base = {"host": "mx", "port": 587} for cfg in ({**base, "from": "alerts@example.com", "username": "u@example.com"}, {**base, "from": "", "username": "u@example.com"}, {**base, "from": "", "username": ""}): - assert (await n._send_email(cfg, ALERT, to))["sent"] + m = message(cfg, to.address, *alert_email(cfg, ALERT, to)) + sent.append((m["From"], m["Subject"])) assert [f for f, _ in sent] == ["alerts@example.com", "u@example.com", "steward@localhost"] # The rule name has a line break; the subject folds it to a space rather than failing. assert sent[0][1] == '[Steward] Firing: Disk "root" on web01 \\ backslash — web01 94%' diff --git a/tests/core/test_outbox.py b/tests/core/test_outbox.py index 792ac77..b1a0cfd 100644 --- a/tests/core/test_outbox.py +++ b/tests/core/test_outbox.py @@ -186,3 +186,40 @@ def test_several_alerts_are_one_email_listing_each_oldest_first(): assert f"Agent offline: {host}" in text and f"Agent offline: {host}" in html assert text.index("web01") < text.index("web02") < text.index("db01") assert "Fri 9 Oct 2026 08:00 EDT" in text # the reader's zone + + +def test_said_names_who_got_it_who_refused_and_who_is_still_being_tried(): + rows = [_row(recipient="a@x", state="sent"), + _row(recipient="b@x", state="failed", last_error="550 no such user"), + _row(recipient="c@x", state="pending", last_error="451 greylisted")] + ok, said = outbox.said(rows) + assert not ok + assert said == ("Sent to a@x. Not sent to b@x: 550 no such user. Not sent yet to c@x " + "(451 greylisted); Steward will try again for up to an hour.") + assert outbox.said(rows[:1]) == (True, "Sent to a@x.") + assert outbox.said([])[0] is False + + +# The guard: every email goes through the queue (System 156 ruling), so the +# only code that opens a connection to the mail server is the outbox's. +def _smtp_openers(sources: dict[str, str]) -> set[str]: + import re + opens = re.compile(r"\bsmtp_open\(|\bsmtplib\.SMTP(_SSL)?\(") + return {path for path, src in sources.items() + if any(opens.search(line) and not line.lstrip().startswith(("#", "def ")) + for line in src.splitlines())} + + +def test_only_the_outbox_talks_to_the_mail_server(): + from pathlib import Path + root = Path(__file__).resolve().parents[2] # the checkout + sources = {str(p.relative_to(root)): p.read_text() for p in (root / "steward").rglob("*.py")} + sources.update({str(p.relative_to(root)): p.read_text() for p in (root / "plugins").rglob("*.py")} + if (root / "plugins").is_dir() else {}) + assert _smtp_openers(sources) == {"steward/core/outbox.py", "steward/core/notifications.py"} + + +def test_the_smtp_guard_can_fail(): + assert _smtp_openers({"x.py": " with smtplib.SMTP(host) as s:"}) == {"x.py"} + assert _smtp_openers({"y.py": " smtp = notifications.smtp_open(cfg)"}) == {"y.py"} + assert _smtp_openers({"z.py": "# smtp_open(cfg) is the outbox's\ndef smtp_open(cfg):"}) == set() diff --git a/tests/integration/_mail.py b/tests/integration/_mail.py new file mode 100644 index 0000000..2ebb867 --- /dev/null +++ b/tests/integration/_mail.py @@ -0,0 +1,29 @@ +"""Email in integration tests: every email is a queued row the outbox sends +(#5506), so a test catches what the mail server would have been handed by +replacing the one SMTP conversation, outbox.deliver_sync.""" +from __future__ import annotations + +from email.message import EmailMessage + + +def capture(app, monkeypatch, answer=None) -> list[EmailMessage]: + """Set a mail server and collect every message the outbox delivers. + `answer(msg)` is the server's reply to each (None: accepted).""" + from steward.core import outbox + sent: list[EmailMessage] = [] + + def deliver(cfg, messages): + sent.extend(messages) + return [answer(m) if answer else None for m in messages] + + app.config["SMTP"] = {**app.config.get("SMTP", {}), "host": "mx.example", "port": 587, + "from": "steward@example.com"} + monkeypatch.setattr(outbox, "deliver_sync", deliver) + return sent + + +def to(sent: list[EmailMessage], *addresses: str) -> list[EmailMessage]: + """The messages among `sent` addressed to any of `addresses` (other + tests' queued email can go out in the same tick).""" + want = {a.lower() for a in addresses} + return [m for m in sent if m["To"].lower() in want] diff --git a/tests/integration/test_alert_email_outbox.py b/tests/integration/test_alert_email_outbox.py index 17a0e2d..d092d14 100644 --- a/tests/integration/test_alert_email_outbox.py +++ b/tests/integration/test_alert_email_outbox.py @@ -81,7 +81,7 @@ def test_alerts_in_one_window_are_one_email_and_each_event_records_it(app, monke from steward.core import outbox tag = uuid.uuid4().hex[:8] - extra = f"oncall-{tag}@test" + extra = f"oncall-{tag}@example.com" app.config["SMTP"] = {"host": "mx.test", "port": 587, "from": "steward@test", "recipients": [extra], "tz": "America/New_York"} t0 = datetime(2026, 10, 9, 12, 0, tzinfo=timezone.utc) @@ -142,7 +142,7 @@ def test_an_email_whose_alerts_were_deleted_fails_and_says_so(app, monkeypatch): from steward.models.alerts import AlertEvent tag = uuid.uuid4().hex[:8] - extra = f"gone-{tag}@test" + extra = f"gone-{tag}@example.com" app.config["SMTP"] = {"host": "mx.test", "port": 587, "from": "steward@test", "recipients": [extra]} monkeypatch.setattr(outbox, "deliver_sync", lambda cfg, msgs: [None] * len(msgs)) t0 = datetime.now(timezone.utc) @@ -155,8 +155,8 @@ def test_an_email_whose_alerts_were_deleted_fails_and_says_so(app, monkeypatch): async with app.db_sessionmaker() as s: async with s.begin(): await s.execute(delete(AlertEvent).where(AlertEvent.id == ev)) - states = await outbox.attempt(app, [row.id]) - return states[row.id], row.id + done = await outbox.attempt(app, [row.id]) + return done[row.id].state, row.id state, rid = asyncio.run(_go()) assert state == "failed" diff --git a/tests/integration/test_alert_send_test.py b/tests/integration/test_alert_send_test.py index b5d812d..f62e1d5 100644 --- a/tests/integration/test_alert_send_test.py +++ b/tests/integration/test_alert_send_test.py @@ -39,11 +39,8 @@ def test_send_test_reaches_each_channel_and_leaves_the_rule_alone(app, monkeypat from steward.models.audit import AuditEvent from steward.models.users import User, UserRole - emails, hooks = [], [] - - async def fake_email(cfg, alert, to): - emails.append((alert, [r.address for r in to])) - return {"sent": True, "error": None} + from tests.integration._mail import capture + hooks = [] async def fake_hook(cfg, alert): hooks.append(alert) @@ -52,10 +49,9 @@ def test_send_test_reaches_each_channel_and_leaves_the_rule_alone(app, monkeypat async def fake_recipients(db, kind, smtp): return [mail.Recipient("ann@example.com", ZoneInfo("UTC"), False, "ann")] - monkeypatch.setattr(notifications, "_send_email", fake_email) + emails = capture(app, monkeypatch) monkeypatch.setattr(notifications, "_send_webhook", fake_hook) monkeypatch.setattr(mail, "recipients", fake_recipients) - app.config["SMTP"] = {"host": "mx.example", "port": 587, "from": "alerts@example.com"} app.config["WEBHOOK"] = {"url": "https://hook.example/api/1"} async def _go(): @@ -83,12 +79,20 @@ def test_send_test_reaches_each_channel_and_leaves_the_rule_alone(app, monkeypat assert resp.status_code == 200 page = await resp.get_data(as_text=True) assert "alert-error" in page - assert "Email sent to ann@example.com." in page + assert "Email: Sent to ann@example.com." in page assert "Webhook to hook.example failed: connection refused" in page - # What went out is this rule, marked TEST, at its latest reading. - ((alert, to),) = emails - assert to == ["ann@example.com"] and hooks == [alert] + # What went out is this rule, marked TEST, at its latest reading: one + # queued test email (recorded, not retried) and the webhook. + (msg,) = emails + (alert,) = hooks + assert msg["To"] == "ann@example.com" and f"Hot {tag}" in msg["Subject"] + from steward.models.outbox import EmailOutbox + async with app.db_sessionmaker() as s: + row = (await s.execute(select(EmailOutbox).where( + EmailOutbox.recipient == "ann@example.com", EmailOutbox.kind == "test", + EmailOutbox.subject == msg["Subject"]))).scalars().one() + assert (row.state, row.retry, row.attempts) == ("sent", False, 1) assert (alert["rule_name"], alert["state"], alert["resource"], alert["value"]) == ( f"Hot {tag}", "TEST", host, 42.0) diff --git a/tests/integration/test_invites.py b/tests/integration/test_invites.py index 3179225..11d908b 100644 --- a/tests/integration/test_invites.py +++ b/tests/integration/test_invites.py @@ -62,16 +62,9 @@ async def _audits(app, uid): return [(r.action, json.loads(r.detail_json or "{}")) for r in rows] -def _mail(app, monkeypatch): - from steward.core import mail - sent = [] - - async def fake_send(cfg, messages): - sent.extend(messages) - - app.config["SMTP"] = {"host": "mx.example", "port": 587, "from": "steward@example.com"} - monkeypatch.setattr(mail, "send", fake_send) - return sent +def _mail(app, monkeypatch, answer=None): + from tests.integration._mail import capture + return capture(app, monkeypatch, answer) def test_an_invite_is_emailed_and_the_invitee_makes_their_own_account_once(app, monkeypatch): @@ -193,3 +186,27 @@ def test_an_address_with_an_account_is_not_invited_and_only_an_admin_invites(app form={"email": "y@example.com"})).status_code == 403 asyncio.run(_go()) + + +def test_an_invite_the_server_refuses_or_defers_shows_the_link_and_why(app, monkeypatch): + # Sent through the queue (#5508) and tried at once: a refusal says so; a + # "try later" says Steward keeps trying; either way the link is shown. + import smtplib + answers = {} + _mail(app, monkeypatch, lambda m: answers.get(m["To"])) + + async def _go(): + admin, aid, _ = await _admin(app) + refused, later = f"no-{aid[:8]}@example.com", f"later-{aid[:8]}@example.com" + answers[refused] = smtplib.SMTPRecipientsRefused({refused: (550, b"no such user")}) + answers[later] = smtplib.SMTPDataError(451, b"greylisted") + resp = await admin.post("/settings/users/invite", headers=CSRF, + form={"email": refused, "role": "viewer", "days": "7"}) + page = await resp.get_data(as_text=True) + assert f"Sending to {refused} failed (550 no such user)." in page and "/invite/" in page + resp = await admin.post("/settings/users/invite", headers=CSRF, + form={"email": later, "role": "viewer", "days": "7"}) + page = await resp.get_data(as_text=True) + assert "451 greylisted" in page and "keep trying for up to an hour" in page + + asyncio.run(_go()) diff --git a/tests/integration/test_notification_test_send.py b/tests/integration/test_notification_test_send.py index e030b96..427d3fb 100644 --- a/tests/integration/test_notification_test_send.py +++ b/tests/integration/test_notification_test_send.py @@ -90,14 +90,14 @@ def test_test_webhook_sends_the_form_template_and_shows_the_body(app, monkeypatc def test_test_email_uses_the_form_settings_and_names_the_tls_mode(app, monkeypatch): - from steward.core import notifications + from steward.core import outbox sent = [] - async def fake_send(cfg, alert, to): - sent.append((cfg, [r.address for r in to])) - return {"sent": True, "error": None} + def fake_deliver(cfg, messages): + sent.append((cfg, [m["To"] for m in messages])) + return [None] * len(messages) - monkeypatch.setattr(notifications, "_send_email", fake_send) + monkeypatch.setattr(outbox, "deliver_sync", fake_deliver) async def _go(): client, uid = await _admin_client(app) @@ -110,10 +110,18 @@ def test_test_email_uses_the_form_settings_and_names_the_tls_mode(app, monkeypat assert "alert-success" in page assert "Sent to a@example.com, b@example.com via mx.example:465 (implicit TLS)." in page assert "Rule: Test alert" in page - # The admin's own address (uuid@test) isn't one mail can reach, so only the typed ones. + # The admin's own address (uuid@test) isn't one mail can reach, so only the typed ones, + # sent through the queue with the form's unsaved values, recorded, and not retried. assert sent == [({"host": "mx.example", "port": 465, "tls": False, "username": "u", "password": "pw", "from": "alerts@example.com", "base_url": ""}, ["a@example.com", "b@example.com"])] + from sqlalchemy import select + from steward.models.outbox import EmailOutbox + async with app.db_sessionmaker() as s: + rows = (await s.execute(select(EmailOutbox).where( + EmailOutbox.kind == "test", EmailOutbox.recipient.in_(["a@example.com", "b@example.com"])) + .order_by(EmailOutbox.created_at.desc()).limit(2))).scalars().all() + assert {(r.state, r.retry) for r in rows} == {("sent", False)} resp = await client.post("/settings/notifications/test-email", form={ "smtp.host": "", "smtp.port": "587", "smtp.recipients": ""}, headers=csrf) diff --git a/tests/integration/test_passwords.py b/tests/integration/test_passwords.py index d85cfcd..477c47a 100644 --- a/tests/integration/test_passwords.py +++ b/tests/integration/test_passwords.py @@ -123,15 +123,14 @@ _KEY = re.compile(r"/reset/([A-Za-z0-9_-]{20,})") def _mail(app, monkeypatch): - from steward.core import mail - sent = [] + from tests.integration._mail import capture + return capture(app, monkeypatch) - async def fake_send(cfg, messages): - sent.extend(messages) - app.config["SMTP"] = {"host": "mx.example", "port": 587, "from": "steward@example.com"} - monkeypatch.setattr(mail, "send", fake_send) - return sent +async def _deliver(app): + """The outbox sender's next tick: a reset email is queued (#5508).""" + from steward.core import outbox + await outbox.send_due(app) async def _stranger(app): @@ -158,8 +157,11 @@ def test_a_forgotten_password_is_reset_from_the_emailed_link(app, monkeypatch): resp = await guest.post("/forgot", headers=CSRF, form={"name": who}) answers.append((resp.status_code, "on its way" in await resp.get_data(as_text=True))) assert answers == [(200, True)] * 3 - assert [m["To"] for m in sent] == [f"{name}@example.com"] - text = sent[0].get_body(("plain",)).get_content() + await _deliver(app) + from tests.integration._mail import to + mine = to(sent, f"{name}@example.com", f"{sso_name}@example.com", f"nobody-{uid[:8]}@example.com") + assert [m["To"] for m in mine] == [f"{name}@example.com"] + text = mine[0].get_body(("plain",)).get_content() assert name in text and "BST" in text key = _KEY.search(text).group(1) @@ -197,7 +199,10 @@ def test_asking_again_kills_the_first_link_and_an_old_link_expires(app, monkeypa guest = await _stranger(app) for _ in range(2): await guest.post("/forgot", headers=CSRF, form={"name": name}) - first, second = (_KEY.search(m.get_body(("plain",)).get_content()).group(1) for m in sent) + await _deliver(app) + from tests.integration._mail import to + first, second = (_KEY.search(m.get_body(("plain",)).get_content()).group(1) + for m in to(sent, f"{name}@example.com")) assert (await guest.get(f"/reset/{first}")).status_code == 404 assert (await guest.get(f"/reset/{second}")).status_code == 200