""" MeshBay Hub — email sending via localhost Postfix. Postfix listens on loopback only (inet_interfaces = loopback-only), so no authentication is needed. See docs/MAIL-SERVER.md for the full setup. **No authentication is needed** is exactly why everything below exists. The hub can emit mail from its own domain to anywhere, and three API paths reach that ability — two of them at an address the caller types. Unbounded, that is an open relay wearing the instance's reputation, so the bounds are here, in the one function every send passes through, rather than at the call sites where the next one added would forget them (`AV9`–`AV10`, §13.5b). """ import asyncio import hashlib import logging import smtplib from datetime import UTC, datetime, timedelta from email.message import EmailMessage log = logging.getLogger(__name__) _hub_domain: str = "meshbay.org" _hub_url: str = "https://meshbay.org" def configure(hub_id: str) -> None: global _hub_domain, _hub_url _hub_domain = hub_id _hub_url = f"https://{hub_id}" class MailRefused(Exception): """The gate below declined to send. Never carries the address.""" # The complete list of reasons this hub will ever send mail. A `send_*` # function that names anything else does not send — and one that names nothing # is a TypeError, because `purpose` is keyword-required. # # registration confirm an address at sign-up # password_reset a code to the address already on file for that account # invite tell a registered member they were invited to a group # invite_link send an invitation link to an address, account or not # email_change confirm a new address before it replaces the old one # # Only `registration`, `email_change` and `invite_link` can reach an address this # hub has no prior relationship with. Both are necessary; both are why the bound that # matters is keyed on the recipient rather than on who asked. ALLOWED_PURPOSES = frozenset({ "registration", "password_reset", "invite", "email_change", "invite_link"}) # The two a person is actively waiting on. They may spend the whole hourly # budget; the other two may not spend the reserved share of it, so a flood of # sign-ups cannot lock out someone trying to recover their passphrase. RECOVERY_PURPOSES = frozenset({"password_reset", "invite"}) # Which messages share a recipient's cooldown. Within a family the cooldown is # shared, so a burst cannot walk around it by switching purpose. Across the two # it is not, because they are two halves of one conversation: an invitation # arrives, the invitee registers a minute later, and the code they asked for # was refused — silently — by the cooldown the invitation had just armed. An # invitation link made that the normal case rather than a rare one. The daily # cap per recipient still counts every message, whichever family. # Someone else writes to you about a group; everything else is a step of your # own account's (a sign-up code, a reset, an address change). _INVITATIONS = frozenset({"invite", "invite_link"}) def _cooldown_family(purpose: str) -> str: return "invitation" if purpose in _INVITATIONS else "account" def destination_key(address: str) -> str: """A stable handle for one recipient that is not the address itself. Hashed because this table would otherwise be the one place in the hub holding a list of plaintext addresses — the rest of the codebase goes to the trouble of encrypting them at rest (S2). """ return "dest:" + hashlib.sha256( address.strip().lower().encode()).hexdigest()[:32] async def _take(db, key: str, window: timedelta, ceiling: int, cooldown: timedelta | None) -> None: """Charge one send against a counter, or raise MailRefused. Rows live in `mail_quota` rather than in a module dict, so the allowance survives the restart that a deploy is. The caller owns the commit — which means a send that then fails to commit is not charged, and that is the right way round: the alternative charges for mail nobody received. """ from meshbay_hub.db.models import MailQuota now = datetime.now(UTC) row = await db.get(MailQuota, key) if row is None: row = MailQuota(key=key, window_start=now, count=0, last_sent=None) db.add(row) started = row.window_start if started.tzinfo is None: started = started.replace(tzinfo=UTC) if now - started >= window: row.window_start, row.count = now, 0 if cooldown is not None and row.last_sent is not None: last = row.last_sent if last.tzinfo is None: last = last.replace(tzinfo=UTC) if now - last < cooldown: raise MailRefused("too soon since the last message to this recipient") if row.count >= ceiling: raise MailRefused("allowance spent for this window") row.count += 1 row.last_sent = now # The two global ceilings, and what each one means to somebody waiting. EXHAUSTED_GENERAL = "general" # sign-ups and address changes refused EXHAUSTED_ALL = "all" # even a passphrase reset is refused async def _announce_exhaustion(scope: str) -> None: """Tell the administrators, once per hour, that the hub stopped sending. In its own session, and never raising: this runs while a request is being refused, and an alert that fails must not turn a refusal into a 500. Same reason `_mark_hosted` opens its own session next door. Once per window, keyed on a row rather than a flag in memory, because the thing being reported is exactly the kind of event a restart would erase — and a hub that is refusing mail is a hub somebody is probably restarting. """ from sqlalchemy import select from meshbay_hub.api.deps import user_is_admin from meshbay_hub.api.notifications import create_notification from meshbay_hub.db.engine import get_session_factory from meshbay_hub.db.models import MailQuota, User try: async with get_session_factory()() as db: key = f"alert:{scope}" now = datetime.now(UTC) row = await db.get(MailQuota, key) if row is not None and row.last_sent is not None: last = row.last_sent if last.tzinfo is None: last = last.replace(tzinfo=UTC) if now - last < timedelta(hours=1): return if row is None: row = MailQuota(key=key, window_start=now, count=0) db.add(row) row.last_sent = now row.count += 1 admins = [u for u in (await db.execute(select(User).where( User.status == "active"))).scalars().all() if user_is_admin(u)] for admin in admins: await create_notification( db, admin.id, f"mail_budget_{scope}", "This hub has stopped sending mail for this hour" if scope == EXHAUSTED_ALL else "This hub has stopped sending sign-up mail for this hour", detail=("Passphrase resets and invitations are refused too." if scope == EXHAUSTED_ALL else "Passphrase resets and invitations still go out."), link="#/admin", aggregate=False, ) await db.commit() log.warning("Mail budget exhausted (%s) — %d administrator(s) told", scope, len(admins)) except Exception as e: log.warning("Could not announce the mail budget: %s", e) async def reserve(db, purpose: str, address: str, *, sender: str = "") -> None: """Charge one send, or raise MailRefused. The caller owns the commit.""" from meshbay_hub import hub_settings if purpose not in ALLOWED_PURPOSES: raise MailRefused(f"not a purpose this hub sends mail for: {purpose!r}") limits = await hub_settings.mail_limits(db) # An invitation link goes to an address that may belong to nobody here, at # the request of anyone who owns a group — which is anyone, since # registration is open. So the account asking is counted too, per day # (docs/MESHBAY_DESIGN.md §3.4, AV29). Charged first: of the three, it is the one # bound on the person causing the mail, and a refusal further down leaves # it spent, which errs towards less mail. No sender, no link mail. if purpose == "invite_link": if not sender: raise MailRefused("an invitation link mail names the account sending it") await _take(db, f"invite_link:{sender}", timedelta(days=1), limits["invite_link_daily_cap"], None) # The instance ceiling first, and the recovery share is subtracted for the # purposes that are not one. Registration is open, so a limit counted per # account or per IP is one an attacker buys more of; this one is not. budget = limits["hourly_budget"] if purpose not in RECOVERY_PURPOSES: budget = max(0, budget - limits["hourly_reserved_for_recovery"]) try: await _take(db, "hour", timedelta(hours=1), budget, None) except MailRefused: # An operator finds out here or not at all: a refusal is otherwise a # line in the journal, and a hub that has stopped sending sign-up # codes looks exactly like one nobody is signing up to. await _announce_exhaustion( EXHAUSTED_ALL if purpose in RECOVERY_PURPOSES else EXHAUSTED_GENERAL) raise # Then the recipient. The cooldown per family (above) first, so a message # refused for coming too soon does not also spend the day's allowance; then # the daily cap, across every purpose, account and endpoint — what a person # being mail-bombed actually experiences, and the only bound that says so. dest = destination_key(address) await _take( db, f"cool:{_cooldown_family(purpose)}:{dest[len('dest:'):]}", timedelta(days=1), 10**9, timedelta(seconds=limits["destination_cooldown_seconds"])) await _take(db, dest, timedelta(days=1), limits["destination_daily_cap"], None) async def status(db) -> dict: """What the operator sees in the panel: is the hub still sending?""" from sqlalchemy import func, select from meshbay_hub import hub_settings from meshbay_hub.db.models import MailQuota limits = await hub_settings.mail_limits(db) row = await db.get(MailQuota, "hour") used = 0 window_start = None if row is not None: started = row.window_start if started.tzinfo is None: started = started.replace(tzinfo=UTC) if datetime.now(UTC) - started < timedelta(hours=1): used, window_start = row.count, started.isoformat() recipients = await db.scalar( select(func.count()).select_from(MailQuota) .where(MailQuota.key.like("dest:%"))) or 0 general = max(0, limits["hourly_budget"] - limits["hourly_reserved_for_recovery"]) # Named rather than left for the reader to compute from two numbers: the # panel draws a warning off these, and "is it still sending" is the # question an operator opens this page to answer. general_exhausted = used >= general all_exhausted = used >= limits["hourly_budget"] return { "general_exhausted": general_exhausted, "all_exhausted": all_exhausted, "hourly_budget": limits["hourly_budget"], "hourly_used": used, "hour_started_at": window_start, "reserved_for_recovery": limits["hourly_reserved_for_recovery"], # What a sign-up would meet right now, which is the number an operator # is actually asking about when they open this panel. "general_remaining": max(0, general - used), "recovery_remaining": max(0, limits["hourly_budget"] - used), "recipients_tracked": recipients, } async def purge_expired_quota(db) -> int: """Drop counters whose window has passed. Returns how many went.""" from sqlalchemy import delete from meshbay_hub.db.models import MailQuota cutoff = datetime.now(UTC) - timedelta(days=1) result = await db.execute( delete(MailQuota).where(MailQuota.window_start < cutoff)) await db.commit() return result.rowcount or 0 def _send(msg: EmailMessage, *, purpose: str) -> bool: """Blocking. Every caller in an async handler must use `send_off_loop`. The gate is here rather than in `send_off_loop` so that it cannot be stepped around: a new `send_*` helper, a script, a test — everything that puts a message on the wire comes through this function, and `purpose` is keyword-required so forgetting it is a TypeError rather than an unrestricted send. """ if purpose not in ALLOWED_PURPOSES: # The structural half of the gate, and the half that needs no state: # it is here rather than only in `reserve` so that nothing which puts a # message on the wire — a new helper, a script, a test — can name a # reason this hub does not send mail for. The counting half is # `reserve`, which needs a database session and so cannot live here. log.warning("Mail refused: %r is not a purpose this hub sends for", purpose) return False try: with smtplib.SMTP("localhost", 25, timeout=10) as s: s.send_message(msg) return True except Exception: log.exception("Failed to send email to %s", _mask_email(msg["To"] or "")) return False async def send_off_loop(db, fn, *args, purpose: str, sender: str = "", **kwargs) -> bool: """Run one of the `send_*` functions below in a worker thread. `smtplib` is synchronous and this one waits up to ten seconds. Called directly from an async handler — which is what all four call sites did — that ten seconds is not one request's, it is **the whole hub's**: no other request is served, no node socket is read, no WebRTC offer is relayed, for as long as the MTA takes to answer. An unreachable mail server made the instance stop responding to everyone, and one of the three paths that reaches it (`PATCH /v1/users/me`) had no rate limit at all. So the cost of a slow MTA is one request now, not the instance. It is also the one door: `reserve` charges the send against the recipient's allowance and the instance's before anything is handed to the thread, and the recipient is `args[0]` because that is the first parameter of every `send_*` function below. Returns whether the message went. The caller owns the commit, so a request that fails afterwards is not charged for mail nobody received. """ try: await reserve(db, purpose, args[0], sender=sender) except MailRefused as refusal: # Never the address: this line goes to the journal. log.warning("Mail refused (%s): %s", purpose, refusal) return False await asyncio.to_thread(fn, *args, **kwargs) return True def send_verification_code(to: str, code: str, recovery_key: str | None = None) -> None: """ Registration verification e-mail. When `recovery_key` is given it is appended to the body so the recipient's mailbox becomes the backup for it (docs/MESHBAY_DESIGN.md §3.6). `recovery_key` is a **pass-through**: it is generated on the client, never stored anywhere on the hub, and never logged — only whether one was present. """ body = ( f"Your verification code is: {code}\n" "\n" "Enter this code to verify your email address.\n" "This code expires in 24 hours.\n" ) if recovery_key: body += ( "\n" "---- Account recovery key ----\n" "\n" "Keep this message. If you ever forget your passphrase, this key is\n" "what restores your access to your groups. It is not stored on the\n" f"server and nobody at {_hub_domain} can recover it for you.\n" "\n" f" {recovery_key}\n" ) body += ( "\n" "If you did not create a MeshBay account, ignore this email.\n" "\n" f"{_hub_url}\n" ) msg = EmailMessage() msg["From"] = f"noreply@{_hub_domain}" msg["To"] = to msg["Subject"] = f"MeshBay — Your verification code: {code}" msg.set_content(body) _send(msg, purpose="registration") log.info("Verification code sent to %s (recovery_key=%s)", _mask_email(to), bool(recovery_key)) def send_email_change_code(to: str, code: str) -> None: msg = EmailMessage() msg["From"] = f"noreply@{_hub_domain}" msg["To"] = to msg["Subject"] = f"MeshBay — Confirm your new email: {code}" msg.set_content( f"Your verification code is: {code}\n" "\n" "Enter this code to confirm your new email address.\n" "This code expires in 24 hours.\n" "\n" "If you did not request this change, ignore this email.\n" "\n" f"{_hub_url}\n" ) _send(msg, purpose="email_change") log.info("Email change code sent to %s", _mask_email(to)) def send_password_reset_code(to: str, code: str) -> None: """ Passphrase-reset code (docs/MESHBAY_DESIGN.md §3.6). This only re-opens hub login; it recovers no group content — that needs the recovery key. """ msg = EmailMessage() msg["From"] = f"noreply@{_hub_domain}" msg["To"] = to msg["Subject"] = f"MeshBay — Passphrase reset code: {code}" msg.set_content( f"Your passphrase reset code is: {code}\n" "\n" "Enter it to set a new passphrase. This code expires in 1 hour.\n" "\n" "This restores your sign-in only. If you also have your recovery key,\n" "you can restore access to your groups in the same step.\n" "\n" "If you did not request this, ignore this email — your account is\n" "unchanged.\n" "\n" f"{_hub_url}\n" ) _send(msg, purpose="password_reset") log.info("Passphrase reset code sent to %s", _mask_email(to)) def send_invite_notification( to: str, code: str, inviter: str, group_name: str, ) -> None: msg = EmailMessage() msg["From"] = f"noreply@{_hub_domain}" msg["To"] = to msg["Subject"] = f"MeshBay — {inviter} invited you to {group_name}" msg.set_content( f"{inviter} invited you to the group \"{group_name}\" on MeshBay.\n" "\n" f"Your one-time code is: {code}\n" "\n" "Open the group and enter this code when prompted.\n" "The code works once and expires in 7 days.\n" "\n" f"{_hub_url}\n" ) _send(msg, purpose="invite") log.info("Invite notification sent to %s", _mask_email(to)) def send_invite_link(to: str, link: str, inviter: str, group_name: str) -> None: """An invitation link, to an address that may have no account yet. Nothing in it comes from the request but the link, and the link is checked by the caller to be this hub's own invitation URL before it gets here. """ msg = EmailMessage() msg["From"] = f"noreply@{_hub_domain}" msg["To"] = to msg["Subject"] = f"MeshBay — {inviter} invited you to {group_name}" msg.set_content( f"{inviter} invited you to the group \"{group_name}\" on MeshBay.\n" "\n" "Open this link to create your account, or to sign in, and join the group:\n" "\n" f"{link}\n" "\n" "It works once, and for seven days.\n" "If you did not expect it, you can ignore this message.\n" ) _send(msg, purpose="invite_link") log.info("Invitation link sent to %s", _mask_email(to)) def hub_url() -> str: """This hub's own address, as every mail it sends names it.""" return _hub_url def _mask_email(email: str) -> str: local, _, domain = email.partition("@") if len(local) <= 2: return f"{'*' * len(local)}@{domain}" return f"{local[0]}{'*' * (len(local) - 2)}{local[-1]}@{domain}"