aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authorChristophe Besson <cbesson@gmail.com>2026-09-12 13:47:49 +0200
committerChristophe Besson <cbesson@gmail.com>2026-09-12 16:36:54 +0200
commite671b931fd594a39fc840916c81b5d4b1f1e3227 (patch)
tree886b1e5eac7d178606510f330b1faa2b4b177892
parent17c06bc23252929af13f4361c4a6300ee76a0c51 (diff)
downloadmeshbay-e671b931fd594a39fc840916c81b5d4b1f1e3227.tar.gz
fix(hub): the mail allowance is written down, and recovery keeps a share
Two dicts in `mail.py` held the budget, so every deploy handed out a fresh one — and this hub is deployed several times a day. A bound a restart forgets is not a bound, for the reason the denylist is persisted rather than held in memory (S3). It is a `mail_quota` table now, one row per counter, the recipient hashed so the table does not become a list of plaintext addresses. The counting moves with it, into an async `reserve` that has a session, and `send_off_loop` is the one door it stands in. `_send` keeps the purpose allow-list: that half needs no state, and it is what stops anything which puts a message on the wire from naming a reason this hub does not send for. The caller owns the commit, so a request that fails afterwards is not charged for mail nobody received. `hourly_reserved_for_recovery` is new. A flood of sign-ups used to be able to spend the whole hour and lock out the person waiting on a passphrase reset; registration and address changes may now spend only the unreserved share. Values changed as agreed: 10 messages a day to one recipient, 300 s between two reset codes. The address-change ceiling and its cooldown were two bounds on one thing — 3 a day and 60 s apart — and collapse into one 48-hour delay. Asking again for the address already pending is exempt: it reaches no new recipient, that recipient is bounded anyway, and without the exemption a typo locked the account out of correcting it for two days. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01T4YmK41VsEURWFdop4EEeT
-rw-r--r--packages/meshbay-hub/src/meshbay_hub/api/groups.py5
-rw-r--r--packages/meshbay-hub/src/meshbay_hub/api/users.py98
-rw-r--r--packages/meshbay-hub/src/meshbay_hub/db/migrations/versions/e5f6a7b8c9d0_add_mail_quota.py36
-rw-r--r--packages/meshbay-hub/src/meshbay_hub/db/models.py22
-rw-r--r--packages/meshbay-hub/src/meshbay_hub/mail.py205
-rw-r--r--packages/meshbay-hub/src/meshbay_hub/tasks/cleanup.py7
-rw-r--r--packages/meshbay-hub/tests/test_mail_is_not_a_relay.py375
7 files changed, 559 insertions, 189 deletions
diff --git a/packages/meshbay-hub/src/meshbay_hub/api/groups.py b/packages/meshbay-hub/src/meshbay_hub/api/groups.py
index c44888a..b903fcf 100644
--- a/packages/meshbay-hub/src/meshbay_hub/api/groups.py
+++ b/packages/meshbay-hub/src/meshbay_hub/api/groups.py
@@ -781,8 +781,9 @@ async def invite_notify(
try:
await mail.send_off_loop(
- mail.send_invite_notification,
- email, body.code, current_user.username, group.name)
+ db, mail.send_invite_notification,
+ email, body.code, current_user.username, group.name,
+ purpose="invite")
except Exception:
return {"status": "send_failed"}
diff --git a/packages/meshbay-hub/src/meshbay_hub/api/users.py b/packages/meshbay-hub/src/meshbay_hub/api/users.py
index 702d687..7cebd91 100644
--- a/packages/meshbay-hub/src/meshbay_hub/api/users.py
+++ b/packages/meshbay-hub/src/meshbay_hub/api/users.py
@@ -14,7 +14,7 @@ from pydantic import BaseModel, field_validator
from sqlalchemy import delete, func, select, update
from sqlalchemy.ext.asyncio import AsyncSession
-from meshbay_hub import mail
+from meshbay_hub import hub_settings, mail
from meshbay_hub.api.deps import get_current_user, require_user_scope
from meshbay_hub.api.middleware import limiter
from meshbay_hub.api.netutil import client_ip
@@ -130,10 +130,6 @@ class RefreshRequest(BaseModel):
# ── Endpoints ─────────────────────────────────────────────────────────────────
-# Between two verification mails to one pending account.
-VERIFICATION_RESEND_COOLDOWN = 120
-
-
@router.post("/register", status_code=201)
@limiter.limit("5/minute")
async def register(
@@ -159,13 +155,16 @@ async def register(
# `mail.py` bounds the recipient regardless, and this stops the
# door being hammered — the answer is the same either way, which
# is the one this endpoint has always given.
+ resend_cooldown = await hub_settings.get_int(
+ db, "mail.verification_resend_cooldown",
+ hub_settings.mail_default("verification_resend_cooldown"))
recent = await db.execute(
select(EmailVerification).where(
EmailVerification.user_id == found.id,
EmailVerification.purpose == "registration",
EmailVerification.created_at
> datetime.now(timezone.utc)
- - timedelta(seconds=VERIFICATION_RESEND_COOLDOWN),
+ - timedelta(seconds=resend_cooldown),
))
if not recent.first():
await _create_and_send_verification(
@@ -245,7 +244,8 @@ async def _create_and_send_verification(
))
await db.flush()
await mail.send_off_loop(
- mail.send_verification_code, email, code, recovery_key=recovery_key)
+ db, mail.send_verification_code, email, code,
+ purpose="registration", recovery_key=recovery_key)
class VerifyEmailRequest(BaseModel):
@@ -657,13 +657,6 @@ class UpdateProfileRequest(BaseModel):
return v
-# One change-of-address mail per account per this many seconds. The window is
-# the account's, not the caller's IP: the point is the mailbox on the receiving
-# end, and an IP is not what fills it.
-EMAIL_CHANGE_COOLDOWN = 60
-MAX_EMAIL_CHANGES_PER_DAY = 3
-
-
@router.patch("/me")
@limiter.limit("10/minute")
async def update_profile(
@@ -688,40 +681,40 @@ async def update_profile(
new_email = body.email.strip()
eh = hash_email_blind(new_email)
- # A per-account floor under the rate limit above, which counts by IP
- # and so is not a bound on how much mail one account can cause.
- recent = await db.execute(
+ # How often one account may point the hub at a *different* address.
+ # Long, because this is the only path where a signed-in account chooses
+ # who receives a message, and a short delay alone still allows one
+ # stranger per interval indefinitely.
+ #
+ # Asking again for the address already pending is exempt: it reaches no
+ # new recipient — and that recipient is bounded by the per-destination
+ # allowance anyway — while without the exemption a typo would lock the
+ # account out of correcting it for the whole window.
+ pending = (await db.execute(
select(EmailVerification).where(
EmailVerification.user_id == current_user.id,
EmailVerification.purpose == "email_change",
- EmailVerification.created_at
- > datetime.now(timezone.utc) - timedelta(seconds=EMAIL_CHANGE_COOLDOWN),
- ))
- if recent.first():
- raise HTTPException(
- status_code=429,
- detail="A code was just sent. Wait a minute before asking again.")
+ EmailVerification.verified_at.is_(None),
+ ).order_by(EmailVerification.created_at.desc()))).scalars().first()
+ same_address_again = pending is not None and pending.email_hash == eh
- # And a ceiling per day, because the cooldown alone still allows one
- # stranger a minute — 1440 a day, from one account, on a hub where
- # registration is open. This is the only path where a signed-in
- # account chooses the recipient, so it is the only one that needs it.
- #
- # Counted from the IP log, not from EmailVerification: the block below
- # deletes this account's unverified rows before writing a new one, so
- # counting those would have counted one, always. (Which is what the
- # first version of this did.) The log entry is worth having for its
- # own sake — this endpoint wrote none, alone among the ones that mail.
- today = await db.execute(
- select(func.count()).select_from(IPLog).where(
- IPLog.user_id == current_user.id,
- IPLog.event == "email_change_request",
- IPLog.timestamp > datetime.now(timezone.utc) - timedelta(days=1),
- ))
- if (today.scalar() or 0) >= MAX_EMAIL_CHANGES_PER_DAY:
- raise HTTPException(
- status_code=429,
- detail="Too many address changes today. Try again tomorrow.")
+ if not same_address_again:
+ cooldown = await hub_settings.get_int(
+ db, "mail.email_change_cooldown",
+ hub_settings.mail_default("email_change_cooldown"))
+ since = datetime.now(timezone.utc) - timedelta(seconds=cooldown)
+ recent = await db.execute(
+ select(IPLog).where(
+ IPLog.user_id == current_user.id,
+ IPLog.event == "email_change_request",
+ IPLog.timestamp > since,
+ ))
+ if recent.first():
+ raise HTTPException(
+ status_code=429,
+ detail="This account changed its address recently. "
+ "Try again later, or ask for a new code for the "
+ "address already pending.")
# Check that no other active/pending account uses this email
dup = await db.execute(
@@ -755,7 +748,8 @@ async def update_profile(
db.add(IPLog(user_id=current_user.id, event="email_change_request",
ip_address=client_ip(request)))
await db.flush()
- await mail.send_off_loop(mail.send_email_change_code, new_email, code)
+ await mail.send_off_loop(db, mail.send_email_change_code, new_email, code,
+ purpose="email_change")
pending_email = new_email
await db.commit()
@@ -930,12 +924,6 @@ class ResetPasswordRequest(BaseModel):
new_auth_key: str
-# One reset mail per account per this many seconds, whoever asks and from
-# wherever. The rate limit above counts by IP, which bounds a caller, not an
-# inbox.
-RESET_MAIL_COOLDOWN = 60
-
-
@router.post("/password/reset-request")
@limiter.limit("5/minute")
async def password_reset_request(
@@ -963,12 +951,15 @@ async def password_reset_request(
# Per account, under the per-IP limit above. Knowing the pair is the
# hard part and this endpoint is careful about it, but once someone
# does, the cost of repeating lands in a mailbox that is not theirs.
+ reset_cooldown = await hub_settings.get_int(
+ db, "mail.reset_cooldown",
+ hub_settings.mail_default("reset_cooldown"))
recent = await db.execute(
select(EmailVerification).where(
EmailVerification.user_id == user.id,
EmailVerification.purpose == "password_reset",
EmailVerification.created_at
- > datetime.now(timezone.utc) - timedelta(seconds=RESET_MAIL_COOLDOWN),
+ > datetime.now(timezone.utc) - timedelta(seconds=reset_cooldown),
))
if recent.first():
return {"status": "sent_if_exists"}
@@ -996,7 +987,8 @@ async def password_reset_request(
await db.flush()
try:
await mail.send_off_loop(
- mail.send_password_reset_code, decrypt_email(user.email), code)
+ db, mail.send_password_reset_code, decrypt_email(user.email), code,
+ purpose="password_reset")
except Exception:
log.exception("Failed to send passphrase reset code")
await db.commit()
diff --git a/packages/meshbay-hub/src/meshbay_hub/db/migrations/versions/e5f6a7b8c9d0_add_mail_quota.py b/packages/meshbay-hub/src/meshbay_hub/db/migrations/versions/e5f6a7b8c9d0_add_mail_quota.py
new file mode 100644
index 0000000..8f2e4d2
--- /dev/null
+++ b/packages/meshbay-hub/src/meshbay_hub/db/migrations/versions/e5f6a7b8c9d0_add_mail_quota.py
@@ -0,0 +1,36 @@
+"""add mail_quota
+
+The hub's outbound mail allowance, kept across restarts. It lived in two
+dicts in `mail.py`, so every deploy handed out a fresh budget — and the hub is
+deployed often. A bound a restart forgets is not a bound.
+
+Revision ID: e5f6a7b8c9d0
+Revises: d4e5f6a7b8c9
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "e5f6a7b8c9d0"
+down_revision: Union[str, Sequence[str], None] = "d4e5f6a7b8c9"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+ op.create_table(
+ "mail_quota",
+ sa.Column("key", sa.String(64), primary_key=True),
+ sa.Column("window_start", sa.DateTime(timezone=True), nullable=False),
+ sa.Column("count", sa.Integer(), nullable=False, server_default="0"),
+ sa.Column("last_sent", sa.DateTime(timezone=True), nullable=True),
+ )
+
+
+def downgrade() -> None:
+ # Dropping this returns the hub to sending with no recorded history, not to
+ # sending without a bound: the limits themselves live in `hub_settings` and
+ # in the configuration file.
+ op.drop_table("mail_quota")
diff --git a/packages/meshbay-hub/src/meshbay_hub/db/models.py b/packages/meshbay-hub/src/meshbay_hub/db/models.py
index e6c7d33..b052e00 100644
--- a/packages/meshbay-hub/src/meshbay_hub/db/models.py
+++ b/packages/meshbay-hub/src/meshbay_hub/db/models.py
@@ -308,6 +308,28 @@ class UserPreference(Base):
value: Mapped[str] = mapped_column(Text, nullable=False)
+class MailQuota(Base):
+ """How much mail has gone where, kept across restarts.
+
+ This was a pair of dicts in `mail.py`, which meant a restart handed out a
+ fresh allowance — and a hub restarts whenever it is deployed. A budget a
+ restart forgets is not a budget, for the same reason the denylist is
+ persisted rather than held in memory (S3).
+
+ One row per thing being counted:
+ `hour` the instance's hourly total
+ `dest:<hash>` one recipient, hashed — this table must not become a
+ list of plaintext addresses (S2)
+ """
+
+ __tablename__ = "mail_quota"
+
+ key: Mapped[str] = mapped_column(String(64), primary_key=True)
+ window_start: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now)
+ count: Mapped[int] = mapped_column(Integer, default=0)
+ last_sent: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+
+
class HubSetting(Base):
"""
Instance-wide settings an admin changes at runtime from the panel.
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:
diff --git a/packages/meshbay-hub/src/meshbay_hub/tasks/cleanup.py b/packages/meshbay-hub/src/meshbay_hub/tasks/cleanup.py
index 7f8a5f2..dfe7c78 100644
--- a/packages/meshbay-hub/src/meshbay_hub/tasks/cleanup.py
+++ b/packages/meshbay-hub/src/meshbay_hub/tasks/cleanup.py
@@ -57,6 +57,13 @@ async def cleanup_loop(get_session):
stale = await purge_stale_pending_users(db)
if stale:
log.info("Purged %d stale pending users", stale)
+ # One row per recipient the hub has written to, and the
+ # window is a day: without this the table grows for the
+ # life of the instance and nothing ever reads the old rows.
+ from meshbay_hub import mail
+ quota = await mail.purge_expired_quota(db)
+ if quota:
+ log.info("Purged %d expired mail counters", quota)
except asyncio.CancelledError:
raise
except Exception as e:
diff --git a/packages/meshbay-hub/tests/test_mail_is_not_a_relay.py b/packages/meshbay-hub/tests/test_mail_is_not_a_relay.py
index a56ab17..7c1a9f4 100644
--- a/packages/meshbay-hub/tests/test_mail_is_not_a_relay.py
+++ b/packages/meshbay-hub/tests/test_mail_is_not_a_relay.py
@@ -55,13 +55,16 @@ def _real_mail_path(monkeypatch):
@pytest.fixture(autouse=True)
-def _fresh_budget():
- """Each test starts with the hub's allowance untouched."""
- mail_mod._destinations.clear()
- mail_mod._hour = (0.0, 0)
- yield
- mail_mod._destinations.clear()
- mail_mod._hour = (0.0, 0)
+def _configured_limits():
+ """What `hub.toml` says, which is what a missing settings row falls back to.
+
+ Set at startup by the lifespan; pinned here so a test that never builds an
+ app still measures against the shipped defaults rather than against zero.
+ """
+ from meshbay_hub import hub_settings
+ from meshbay_hub.config import MailConfig
+
+ hub_settings.set_mail_defaults(MailConfig())
@pytest.fixture
@@ -96,6 +99,17 @@ def _msg(to: str):
return m
+async def _charge(db, purpose: str, address: str) -> bool:
+ """One send against the allowance. True if it was granted."""
+ try:
+ await mail_mod.reserve(db, purpose, address)
+ except mail_mod.MailRefused:
+ await db.rollback()
+ return False
+ await db.commit()
+ return True
+
+
# ── The gate itself ──────────────────────────────────────────────────────────
def test_a_send_must_name_a_purpose(wire):
@@ -105,50 +119,115 @@ def test_a_send_must_name_a_purpose(wire):
_REAL_SEND(_msg("someone@example.test"))
-def test_only_the_four_purposes_send(wire):
- assert _REAL_SEND(_msg("a@example.test"), purpose="registration")
+def test_send_refuses_a_purpose_this_hub_does_not_have(wire):
+ """The structural half of the gate, and the half that needs no state. It
+ is in `_send` so that nothing which puts a message on the wire can name a
+ reason this hub does not send mail for; the counting half needs a database
+ session and lives in `reserve`."""
assert not _REAL_SEND(_msg("b@example.test"), purpose="marketing")
assert not _REAL_SEND(_msg("c@example.test"), purpose="")
+ assert _REAL_SEND(_msg("a@example.test"), purpose="registration")
assert [to for to, _ in wire] == ["a@example.test"]
-def test_one_recipient_has_a_cooldown_and_a_daily_allowance(wire):
- """Across every purpose — the recipient is the thing being protected, not
- the sender."""
+@pytest.mark.asyncio
+async def test_reserve_refuses_a_purpose_this_hub_does_not_have(db_session):
+ with pytest.raises(mail_mod.MailRefused):
+ await mail_mod.reserve(db_session, "marketing", "a@example.test")
+
+
+@pytest.mark.asyncio
+async def test_one_recipient_has_a_cooldown_and_a_daily_allowance(db_session):
+ """Across every purpose — the recipient is what is protected, not the
+ sender."""
+ from meshbay_hub.config import MailConfig
+ from meshbay_hub.db.models import MailQuota
+
victim = "victim@example.test"
- assert _REAL_SEND(_msg(victim), purpose="registration")
- assert not _REAL_SEND(_msg(victim), purpose="password_reset"), (
+ assert await _charge(db_session, "registration", victim)
+ assert not await _charge(db_session, "password_reset", victim), (
"a second purpose walked around the cooldown")
- # Past the cooldown, up to the day's allowance, then no more.
- for _ in range(mail_mod.DESTINATION_DAILY_CAP - 1):
- mail_mod._destinations[mail_mod._destination_key(victim)] = (
- 0.0,
- mail_mod._destinations[mail_mod._destination_key(victim)][1],
- mail_mod._destinations[mail_mod._destination_key(victim)][2])
- assert _REAL_SEND(_msg(victim), purpose="invite")
+ cap = MailConfig().destination_daily_cap
+ key = mail_mod.destination_key(victim)
+ for _ in range(cap - 1):
+ # Step past the cooldown without waiting it out. The daily count is
+ # deliberately left alone.
+ (await db_session.get(MailQuota, key)).last_sent = None
+ await db_session.commit()
+ assert await _charge(db_session, "invite", victim)
- mail_mod._destinations[mail_mod._destination_key(victim)] = (
- 0.0, mail_mod.DESTINATION_DAILY_CAP,
- mail_mod._destinations[mail_mod._destination_key(victim)][2])
- assert not _REAL_SEND(_msg(victim), purpose="invite")
- assert len(wire) == mail_mod.DESTINATION_DAILY_CAP
+ (await db_session.get(MailQuota, key)).last_sent = None
+ await db_session.commit()
+ assert not await _charge(db_session, "invite", victim), (
+ f"the {cap}-a-day allowance for one recipient was not enforced")
-def test_the_same_recipient_is_one_recipient_however_it_is_written(wire):
+@pytest.mark.asyncio
+async def test_the_same_recipient_is_one_recipient_however_it_is_written(db_session):
"""Case and surrounding space are not a new person."""
- assert _REAL_SEND(_msg("Victim@Example.Test"), purpose="registration")
- assert not _REAL_SEND(_msg(" victim@example.test "), purpose="registration")
- assert len(wire) == 1
+ assert await _charge(db_session, "registration", "Victim@Example.Test")
+ assert not await _charge(db_session, "registration", " victim@example.test ")
-def test_the_instance_has_a_ceiling_no_account_can_buy_past(wire):
+@pytest.mark.asyncio
+async def test_the_instance_has_a_ceiling_no_account_can_buy_past(db_session):
"""Registration is open, so "per account" is a bound an attacker buys more
- of. This one cannot be bought."""
- for i in range(mail_mod.HOURLY_BUDGET):
- assert _REAL_SEND(_msg(f"u{i}@example.test"), purpose="registration")
- assert not _REAL_SEND(_msg("one-more@example.test"), purpose="registration")
- assert len(wire) == mail_mod.HOURLY_BUDGET
+ of. This one is not."""
+ from meshbay_hub.config import MailConfig
+ cfg = MailConfig()
+ general = cfg.hourly_budget - cfg.hourly_reserved_for_recovery
+
+ for i in range(general):
+ assert await _charge(db_session, "registration", f"u{i}@example.test")
+ assert not await _charge(db_session, "registration", "one-more@example.test")
+
+
+@pytest.mark.asyncio
+async def test_a_flood_of_sign_ups_cannot_lock_out_a_passphrase_reset(db_session):
+ """The reserved share: someone waiting on a reset code does not wait
+ because strangers were registering all hour."""
+ from meshbay_hub.config import MailConfig
+ cfg = MailConfig()
+ general = cfg.hourly_budget - cfg.hourly_reserved_for_recovery
+
+ for i in range(general):
+ assert await _charge(db_session, "registration", f"flood{i}@example.test")
+ assert not await _charge(db_session, "registration", "another@example.test")
+
+ assert await _charge(db_session, "password_reset", "waiting@example.test"), (
+ "a sign-up flood spent the share kept for recovery")
+ assert await _charge(db_session, "invite", "invited@example.test")
+
+
+@pytest.mark.asyncio
+async def test_the_allowance_is_written_down_and_survives_a_restart(db_session):
+ """It was two module dicts, so every deploy handed out a fresh budget —
+ and the hub is deployed often."""
+ from meshbay_hub.db.models import MailQuota
+
+ assert await _charge(db_session, "registration", "kept@example.test")
+
+ # A restart loses everything that was not written down. These were.
+ assert await db_session.get(MailQuota, "hour") is not None
+ assert await db_session.get(
+ MailQuota, mail_mod.destination_key("kept@example.test")) is not None
+ assert not await _charge(db_session, "registration", "kept@example.test")
+
+
+@pytest.mark.asyncio
+async def test_a_bound_an_admin_changed_is_the_one_that_applies(db_session):
+ """These are settings, not constants: the hour a budget runs out is not
+ when anyone wants to edit a file and restart."""
+ from meshbay_hub import hub_settings
+
+ await hub_settings.set_raw(db_session, "mail.destination_cooldown_seconds", "0")
+ await hub_settings.set_raw(db_session, "mail.destination_daily_cap", "2")
+ await db_session.commit()
+
+ assert await _charge(db_session, "registration", "tuned@example.test")
+ assert await _charge(db_session, "registration", "tuned@example.test")
+ assert not await _charge(db_session, "registration", "tuned@example.test")
def test_a_refusal_never_names_the_address():
@@ -168,11 +247,6 @@ def test_a_refusal_never_names_the_address():
def emit(self, record):
records.append(record.getMessage())
- # The refusal itself, which is what the log line is built from.
- with pytest.raises(mail_mod.MailRefused) as refusal:
- mail_mod._budget_or_refuse("marketing", "private@example.test")
- assert "private@example.test" not in str(refusal.value)
-
# And the line as emitted. Level and `logging.disable` are both forced:
# attaching a handler is not enough when an earlier test has raised the
# logger's level, which is why the first version of this saw nothing in
@@ -262,52 +336,209 @@ async def _signed_in(client, db_session, username: str, email: str) -> dict:
@pytest.mark.asyncio
-async def test_a_signed_in_account_cannot_mail_a_stranger_repeatedly(
- client, db_session, wire, monkeypatch):
+async def test_an_account_may_point_the_hub_at_one_stranger_then_wait(
+ client, db_session, wire):
"""
- `PATCH /v1/users/me` is the one path where a signed-in account chooses the
- recipient. A cooldown alone still allowed one stranger a minute — 1440 a
- day — so there is a ceiling per day above it.
+ `PATCH /v1/users/me` is the only path where a signed-in account chooses the
+ recipient. A short delay alone still allows one stranger per interval,
+ indefinitely, so the delay is long — and a second, different address
+ inside it is refused.
"""
- monkeypatch.setattr(
- "meshbay_hub.api.users.EMAIL_CHANGE_COOLDOWN", 0)
-
headers = await _signed_in(client, db_session, "relay_b",
"relay_b@example.test")
before = len(wire)
- accepted = 0
- for i in range(8):
- r = await client.patch("/v1/users/me", headers=headers,
- json={"email": f"stranger{i}@example.test"})
- if r.status_code == 200:
- accepted += 1
- else:
- assert r.status_code == 429, r.text
+ r = await client.patch("/v1/users/me", headers=headers,
+ json={"email": "a-stranger@example.test"})
+ assert r.status_code == 200, r.text
+
+ r = await client.patch("/v1/users/me", headers=headers,
+ json={"email": "another-stranger@example.test"})
+ assert r.status_code == 429, r.text
+ assert len(wire) - before == 1, "the hub mailed a second stranger on demand"
+
+
+@pytest.mark.asyncio
+async def test_the_address_already_pending_may_be_asked_for_again(
+ client, db_session, wire):
+ """
+ The exemption, and why it is not a hole: re-asking for a code for the
+ address already pending reaches no new recipient, and that recipient is
+ bounded by its own daily allowance regardless. Without it, a typo would
+ lock the account out of correcting it for the whole window.
+ """
+ from meshbay_hub import hub_settings
+ from meshbay_hub.db.models import MailQuota
+
+ headers = await _signed_in(client, db_session, "relay_d",
+ "relay_d@example.test")
+ typo = "jean@gmial.test"
- from meshbay_hub.api.users import MAX_EMAIL_CHANGES_PER_DAY
- assert accepted == MAX_EMAIL_CHANGES_PER_DAY, (
- f"{accepted} strangers mailed by one account in one day")
- assert len(wire) - before == MAX_EMAIL_CHANGES_PER_DAY
+ r = await client.patch("/v1/users/me", headers=headers, json={"email": typo})
+ assert r.status_code == 200, r.text
+
+ # The recipient's own cooldown is not what is under test here.
+ await hub_settings.set_raw(db_session, "mail.destination_cooldown_seconds", "0")
+ await db_session.commit()
+ row = await db_session.get(MailQuota, mail_mod.destination_key(typo))
+ row.last_sent = None
+ await db_session.commit()
+
+ before = len(wire)
+ r = await client.patch("/v1/users/me", headers=headers, json={"email": typo})
+ assert r.status_code == 200, (
+ "a typo locked the account out of correcting it: " + r.text)
+ assert len(wire) - before == 1
@pytest.mark.asyncio
-async def test_the_daily_ceiling_survives_the_row_being_deleted(
- client, db_session, wire, monkeypatch):
+async def test_the_delay_survives_the_verification_row_being_deleted(
+ client, db_session, wire):
"""
- The change path deletes this account's unverified verification rows before
- writing a new one, so a ceiling counted from that table counts one, always
- — which is what the first version of this counted. It comes off the IP log.
+ The handler deletes this account's unverified rows before writing a new
+ one, so a window counted from that table would count one, always. Which is
+ what the first version of this counted. It comes off the IP log.
"""
- monkeypatch.setattr("meshbay_hub.api.users.EMAIL_CHANGE_COOLDOWN", 0)
headers = await _signed_in(client, db_session, "relay_c",
"relay_c@example.test")
- for i in range(3):
- await client.patch("/v1/users/me", headers=headers,
- json={"email": f"c-stranger{i}@example.test"})
+ r = await client.patch("/v1/users/me", headers=headers,
+ json={"email": "c-first@example.test"})
+ assert r.status_code == 200, r.text
+
+ from sqlalchemy import delete
+
+ from meshbay_hub.db.models import EmailVerification
+ await db_session.execute(delete(EmailVerification))
+ await db_session.commit()
r = await client.patch("/v1/users/me", headers=headers,
- json={"email": "c-one-more@example.test"})
+ json={"email": "c-second@example.test"})
assert r.status_code == 429, (
- "the ceiling was counted from a table the handler empties")
+ "the delay was counted from a table the handler empties")
+
+
+# ── The operator's controls ──────────────────────────────────────────────────
+
+async def _admin(client, db_session, username: str) -> dict:
+ from sqlalchemy import update
+
+ from meshbay_hub.db.models import User
+
+ headers = await _signed_in(client, db_session, username,
+ f"{username}@example.test")
+ await db_session.execute(
+ update(User).where(User.username == username).values(role="admin"))
+ await db_session.commit()
+ return headers
+
+
+@pytest.mark.asyncio
+async def test_the_panel_shows_the_bounds_their_defaults_and_the_range(
+ client, db_session):
+ """The panel draws the fields from this, so a value it cannot store is one
+ the operator is told about rather than one that is silently clamped."""
+ headers = await _admin(client, db_session, "mailadmin1")
+
+ r = await client.get("/v1/admin/settings", headers=headers)
+ assert r.status_code == 200, r.text
+ body = r.json()
+ from meshbay_hub import hub_settings
+ assert set(body["mail"]) == set(hub_settings.MAIL_KEYS)
+ assert set(body["mail_defaults"]) == set(hub_settings.MAIL_KEYS)
+ assert body["mail_bounds"]["hourly_budget"][0] >= 1, (
+ "a budget of zero would stop the hub sending anything at all")
+
+
+@pytest.mark.asyncio
+async def test_an_admin_changes_a_bound_and_it_takes_effect(client, db_session):
+ headers = await _admin(client, db_session, "mailadmin2")
+
+ r = await client.patch("/v1/admin/settings", headers=headers,
+ json={"mail": {"destination_daily_cap": 3}})
+ assert r.status_code == 200, r.text
+ assert r.json()["mail"]["destination_daily_cap"] == 3
+
+ from meshbay_hub import hub_settings
+ assert (await hub_settings.mail_limits(db_session))[
+ "destination_daily_cap"] == 3
+
+
+@pytest.mark.asyncio
+async def test_a_value_outside_the_range_is_clamped_not_stored(client, db_session):
+ """A number typed into a web form is not a reason to send without a bound."""
+ headers = await _admin(client, db_session, "mailadmin3")
+
+ r = await client.patch("/v1/admin/settings", headers=headers,
+ json={"mail": {"hourly_budget": 0,
+ "destination_daily_cap": 10 ** 9}})
+ assert r.status_code == 200, r.text
+ from meshbay_hub import hub_settings
+ low, _ = hub_settings.MAIL_BOUNDS["hourly_budget"]
+ _, high = hub_settings.MAIL_BOUNDS["destination_daily_cap"]
+ assert r.json()["mail"]["hourly_budget"] == low
+ assert r.json()["mail"]["destination_daily_cap"] == high
+
+
+@pytest.mark.asyncio
+async def test_an_unknown_mail_setting_is_refused(client, db_session):
+ headers = await _admin(client, db_session, "mailadmin4")
+ r = await client.patch("/v1/admin/settings", headers=headers,
+ json={"mail": {"send_to_everyone": 1}})
+ assert r.status_code == 422, r.text
+
+
+@pytest.mark.asyncio
+async def test_a_moderator_may_read_the_bounds_but_not_change_them(
+ client, db_session):
+ from sqlalchemy import update
+
+ from meshbay_hub.db.models import User
+
+ headers = await _signed_in(client, db_session, "mailmod",
+ "mailmod@example.test")
+ await db_session.execute(
+ update(User).where(User.username == "mailmod").values(role="moderator"))
+ await db_session.commit()
+
+ assert (await client.get("/v1/admin/settings", headers=headers)).status_code == 200
+ r = await client.patch("/v1/admin/settings", headers=headers,
+ json={"mail": {"hourly_budget": 10}})
+ assert r.status_code == 403, r.text
+
+
+@pytest.mark.asyncio
+async def test_the_operator_can_see_whether_the_hub_is_still_sending(
+ client, db_session, wire):
+ """There was no way to see this at all: a refusal was a line in the
+ journal, so an instance that had stopped sending sign-up codes looked,
+ from the panel, exactly like one with no sign-ups."""
+ headers = await _admin(client, db_session, "mailadmin5")
+
+ r = await client.get("/v1/admin/mail", headers=headers)
+ assert r.status_code == 200, r.text
+ before = r.json()
+ # A delta, not an absolute: creating the admin above signed somebody up,
+ # which sent a verification code and spent one of the hour's allowance.
+ assert before["general_remaining"] < before["recovery_remaining"], (
+ "the reserved share must be visible as the difference between the two")
+
+ assert await _charge(db_session, "registration", "seen@example.test")
+
+ after = (await client.get("/v1/admin/mail", headers=headers)).json()
+ assert after["hourly_used"] == before["hourly_used"] + 1
+ assert after["general_remaining"] == before["general_remaining"] - 1
+ assert after["recipients_tracked"] > before["recipients_tracked"]
+
+
+def test_the_hub_refuses_to_start_with_more_than_one_worker():
+ """The node registry and the signaling futures are per-process dicts, so a
+ second worker makes a node intermittently unreachable for half its
+ members — a symptom that describes something else entirely."""
+ from meshbay_hub.daemon import single_worker_or_exit
+
+ single_worker_or_exit(1)
+ for workers in (0, 2, 8):
+ with pytest.raises(SystemExit) as exit_:
+ single_worker_or_exit(workers)
+ assert exit_.value.code == 2