diff options
Diffstat (limited to 'packages/meshbay-hub/src/meshbay_hub/mail.py')
| -rw-r--r-- | packages/meshbay-hub/src/meshbay_hub/mail.py | 205 |
1 files changed, 143 insertions, 62 deletions
diff --git a/packages/meshbay-hub/src/meshbay_hub/mail.py b/packages/meshbay-hub/src/meshbay_hub/mail.py index 5d13b8d..661a1f3 100644 --- a/packages/meshbay-hub/src/meshbay_hub/mail.py +++ b/packages/meshbay-hub/src/meshbay_hub/mail.py @@ -16,12 +16,22 @@ import asyncio import hashlib import logging import smtplib -import time +from datetime import datetime, timedelta, timezone 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.""" @@ -36,82 +46,136 @@ class MailRefused(Exception): # email_change confirm a new address before it replaces the old one # # Only `registration` and `email_change` can reach an address this hub has no -# prior relationship with. Both are necessary; both are why the per-destination -# bound below is keyed on the recipient rather than on who asked. +# 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"}) -# Per recipient, across every purpose, every account and every IP. This is the -# bound that matters: it is what a person being mail-bombed actually -# experiences, and no combination of accounts, addresses or endpoints moves it. -DESTINATION_COOLDOWN_SECONDS = 120 -DESTINATION_DAILY_CAP = 5 +# 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"}) -# Instance-wide ceiling. Registration is open, so "per account" is a bound an -# attacker buys more of; this one cannot be bought. Sized far above a real -# hub's traffic — meshbay.org sends single-digit mails a day — and low enough -# that being used as a relay is not worth the trouble. -HOURLY_BUDGET = 200 -# Single-process state, like `_connected_nodes` and `_node_groups` next door: -# the hub serves on one worker (`server.workers` defaults to 1) and the -# signaling registries already require it. A restart clears these, which costs -# at most one burst and is not a security decision — unlike the denylist, -# which persists for exactly that reason (S3). -_destinations: dict[str, tuple[float, int, float]] = {} # key → (last, day count, day start) -_hour: tuple[float, int] = (0.0, 0) # (window start, count) +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] -def _destination_key(address: str) -> str: - """A stable handle for one recipient that is not the address itself. - Hashed because this dict is the one place in the hub that would otherwise - hold a list of plaintext addresses in memory — the rest of the codebase - goes to the trouble of encrypting them at rest (S2). +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. """ - return hashlib.sha256(address.strip().lower().encode()).hexdigest()[:32] + from meshbay_hub.db.models import MailQuota + + now = datetime.now(timezone.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=timezone.utc) + if now - started >= window: + row.window_start, row.count = now, 0 -def _budget_or_refuse(purpose: str, address: str) -> None: - """Raise MailRefused unless this hub may send this, to this person, now.""" - global _hour + 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=timezone.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 + + +async def reserve(db, purpose: str, address: 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}") - now = time.monotonic() - start, count = _hour - if now - start >= 3600: - start, count = now, 0 - if count >= HOURLY_BUDGET: - _hour = (start, count) - raise MailRefused("the hub's hourly mail budget is spent") + limits = await hub_settings.mail_limits(db) - key = _destination_key(address) - last, sent_today, day_start = _destinations.get(key, (0.0, 0, now)) - if now - day_start >= 86400: - sent_today, day_start = 0, now - if last and now - last < DESTINATION_COOLDOWN_SECONDS: - raise MailRefused("too soon since the last mail to this recipient") - if sent_today >= DESTINATION_DAILY_CAP: - raise MailRefused("this recipient has had its daily allowance") + # 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"]) + await _take(db, "hour", timedelta(hours=1), budget, None) - if len(_destinations) > 10_000: - for k, (_, _, started) in list(_destinations.items()): - if now - started >= 86400: - _destinations.pop(k, None) + # Then the recipient: across every purpose, account and endpoint. This is + # what a person being mail-bombed actually experiences, and the only bound + # that describes it. + await _take( + db, destination_key(address), timedelta(days=1), + limits["destination_daily_cap"], + timedelta(seconds=limits["destination_cooldown_seconds"])) - _destinations[key] = (now, sent_today + 1, day_start) - _hour = (start, count + 1) -_hub_domain: str = "meshbay.org" -_hub_url: str = "https://meshbay.org" +async def status(db) -> dict: + """What the operator sees in the panel: is the hub still sending?""" + from meshbay_hub import hub_settings + from meshbay_hub.db.models import MailQuota + from sqlalchemy import func, select + 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=timezone.utc) + if datetime.now(timezone.utc) - started < timedelta(hours=1): + used, window_start = row.count, started.isoformat() -def configure(hub_id: str) -> None: - global _hub_domain, _hub_url - _hub_domain = hub_id - _hub_url = f"https://{hub_id}" + 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"]) + return { + "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 meshbay_hub.db.models import MailQuota + from sqlalchemy import delete + + cutoff = datetime.now(timezone.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: @@ -123,11 +187,14 @@ def _send(msg: EmailMessage, *, purpose: str) -> bool: keyword-required so forgetting it is a TypeError rather than an unrestricted send. """ - try: - _budget_or_refuse(purpose, msg["To"] or "") - except MailRefused as refusal: - # Never the address: this line goes to the journal. - log.warning("Mail refused (%s): %s", purpose, refusal) + 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: @@ -138,7 +205,7 @@ def _send(msg: EmailMessage, *, purpose: str) -> bool: return False -async def send_off_loop(fn, *args, **kwargs) -> None: +async def send_off_loop(db, fn, *args, purpose: 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 @@ -150,8 +217,22 @@ async def send_off_loop(fn, *args, **kwargs) -> None: 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]) + 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: |