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

1 files changed, 45 insertions, 53 deletions

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()