fix(hub): no mail from the event loop, and a ceiling on every path that sends it

`users.py` and `admin.py` under the availability lens. `admin.py` needed
nothing — its moderator/admin line is drawn explicitly, self-modification is
refused, and every list it serves is bounded. `users.py` had four findings and
one of them is the worst of this whole pass.

  AV9   `mail._send` is `smtplib` with a ten-second timeout, called straight
        from four async handlers. That wait is not one request's, it is the
        instance's: nothing else served, no node socket read, no WebRTC offer
        relayed, until the MTA answers. Reachable by any signed-in user at
        request rate through the endpoint below. It has no symptom a test
        catches — everything simply works slowly, for everyone, whenever the
        mail server is having a bad day.

  AV10  `PATCH /v1/users/me` is the third path that makes the hub send mail
        and the only one with neither a rate limit nor a captcha, while
        `register` and `password/reset-request` have both. The address is any
        string the caller types and the duplicate check only rejects one
        already held by an account here, so every address *not* registered on
        this hub was a valid target: a relay for verification codes with the
        hub's own reputation attached. A rate limit counting by IP bounds a
        caller and not an inbox, so the floor under it is a cooldown per
        account — the same for a reset request, whose cost also lands in a
        mailbox that is not the asker's.

  AV11  `default_tab:` accepted any suffix on a `{key:path}` route with an
        unbounded Text value and no cap on rows: one account could write
        without limit into a table shared with everyone. The suffix is a group
        id, which is what the SPA writes, so it is checked as one. A key over
        64 characters was also a 500 rather than a 400 — the column is
        String(64), which PostgreSQL enforces and SQLite does not, so it would
        have appeared in production and in no test.

  AV12  `/v1/notifications` and `/v1/groups` had no upper bound on `limit` and
        no floor under `offset`, while every list in `admin.py` carries
        `le=200`. The group directory takes no authentication at all.

Two shapes recur and are now named in §13.5b: a limit written on one of
several equivalent paths, and a bound that counts the wrong thing.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01T4YmK41VsEURWFdop4EEeT

3 files changed, 101 insertions, 16 deletions

diff --git a/packages/meshbay-hub/src/meshbay_hub/api/groups.py b/packages/meshbay-hub/src/meshbay_hub/api/groups.py
index b1a22f7..c44888a 100644
--- a/packages/meshbay-hub/src/meshbay_hub/api/groups.py
+++ b/packages/meshbay-hub/src/meshbay_hub/api/groups.py
@@ -2,7 +2,7 @@
 
 import re
 
-from fastapi import APIRouter, Depends, HTTPException, Request
+from fastapi import APIRouter, Depends, HTTPException, Query, Request
 from pydantic import BaseModel
 from datetime import datetime, timezone
 from sqlalchemy import func, or_, select, update
@@ -140,9 +140,12 @@ async def group_online_nodes(
 @router.get("")
 async def list_public_groups(
     db: AsyncSession = Depends(get_db),
-    q: str = "",
-    limit: int = 50,
-    offset: int = 0,
+    q: str = Query(default="", max_length=200),
+    # This one takes no authentication at all, and had no upper bound: any
+    # stranger could ask the hub for the entire public directory in a single
+    # query, repeatedly. Bounded like every list in admin.py.
+    limit: int = Query(default=50, ge=1, le=200),
+    offset: int = Query(default=0, ge=0),
     include_federated: bool = True,
 ):
     """List/search public groups — local and optionally federated. No auth required."""
@@ -777,7 +780,8 @@ async def invite_notify(
         return {"status": "no_email"}
 
     try:
-        mail.send_invite_notification(
+        await mail.send_off_loop(
+            mail.send_invite_notification,
             email, body.code, current_user.username, group.name)
     except Exception:
         return {"status": "send_failed"}
diff --git a/packages/meshbay-hub/src/meshbay_hub/api/notifications.py b/packages/meshbay-hub/src/meshbay_hub/api/notifications.py
index 9d5c125..b5783ab 100644
--- a/packages/meshbay-hub/src/meshbay_hub/api/notifications.py
+++ b/packages/meshbay-hub/src/meshbay_hub/api/notifications.py
@@ -19,7 +19,7 @@ migration for no gain, and `unread_only` stays because it is what an older
 interface asks for and it still answers correctly — every row is unread.
 """
 
-from fastapi import APIRouter, Depends, HTTPException
+from fastapi import APIRouter, Depends, HTTPException, Query
 from datetime import datetime, timezone
 
 from sqlalchemy import delete, func, select
@@ -36,8 +36,11 @@ router = APIRouter(prefix="/v1/notifications", tags=["notifications"])
 async def list_notifications(
     current_user: User = Depends(get_current_user),
     db: AsyncSession = Depends(get_db),
-    limit: int = 50,
-    offset: int = 0,
+    # Bounded like every list in admin.py. These two were not, so one caller
+    # could ask for the whole table in one query — and a negative limit is a
+    # 500 on PostgreSQL rather than an empty page.
+    limit: int = Query(default=50, ge=1, le=200),
+    offset: int = Query(default=0, ge=0),
     unread_only: bool = False,
 ):
     query = select(Notification).where(Notification.user_id == current_user.id)
diff --git a/packages/meshbay-hub/src/meshbay_hub/api/users.py b/packages/meshbay-hub/src/meshbay_hub/api/users.py
index a1b436e..05bf075 100644
--- a/packages/meshbay-hub/src/meshbay_hub/api/users.py
+++ b/packages/meshbay-hub/src/meshbay_hub/api/users.py
@@ -2,6 +2,7 @@
 
 import base64
 import logging
+import re
 import secrets
 import time
 import uuid
@@ -10,7 +11,7 @@ from datetime import datetime, timedelta, timezone
 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
 from fastapi import APIRouter, Depends, HTTPException, Request
 from pydantic import BaseModel, field_validator
-from sqlalchemy import delete, select, update
+from sqlalchemy import delete, func, select, update
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from meshbay_hub import mail
@@ -222,7 +223,8 @@ async def _create_and_send_verification(
         expires_at=datetime.now(timezone.utc) + timedelta(seconds=VERIFICATION_TTL),
     ))
     await db.flush()
-    mail.send_verification_code(email, code, recovery_key=recovery_key)
+    await mail.send_off_loop(
+        mail.send_verification_code, email, code, recovery_key=recovery_key)
 
 
 class VerifyEmailRequest(BaseModel):
@@ -634,17 +636,50 @@ 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
+
+
 @router.patch("/me")
+@limiter.limit("10/minute")
 async def update_profile(
     body: UpdateProfileRequest,
+    request: Request,
     current_user: User = Depends(require_user_scope),
     db: AsyncSession = Depends(get_db),
 ):
+    """Update the signed-in account. Changing the address sends a code to it.
+
+    Availability: this is the third path that makes the hub send mail, and it
+    was the one with no rate limit and no captcha — while `register` and
+    `password/reset-request` have both. The address is any string the caller
+    types, and the duplicate check below only rejects one already held by an
+    account here, so every address that is *not* registered on this hub was a
+    valid target. Any signed-in user could therefore have the hub mail
+    arbitrary strangers, from its own domain, as fast as it would go: a relay
+    for verification codes with someone else's reputation attached.
+    """
     pending_email = None
     if body.email is not None:
         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(
+            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.")
+
         # Check that no other active/pending account uses this email
         dup = await db.execute(
             select(User).where(
@@ -675,7 +710,7 @@ async def update_profile(
             expires_at=datetime.now(timezone.utc) + timedelta(seconds=VERIFICATION_TTL),
         ))
         await db.flush()
-        mail.send_email_change_code(new_email, code)
+        await mail.send_off_loop(mail.send_email_change_code, new_email, code)
         pending_email = new_email
 
     await db.commit()
@@ -850,6 +885,12 @@ 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(
@@ -874,6 +915,19 @@ async def password_reset_request(
     )
 
     if matched:
+        # 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.
+        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),
+            ))
+        if recent.first():
+            return {"status": "sent_if_exists"}
+
         prev = await db.execute(
             select(EmailVerification).where(
                 EmailVerification.user_id == user.id,
@@ -896,7 +950,8 @@ async def password_reset_request(
                      ip_address=client_ip(request)))
         await db.flush()
         try:
-            mail.send_password_reset_code(decrypt_email(user.email), code)
+            await mail.send_off_loop(
+                mail.send_password_reset_code, decrypt_email(user.email), code)
         except Exception:
             log.exception("Failed to send passphrase reset code")
         await db.commit()
@@ -974,12 +1029,27 @@ ALLOWED_PREF_KEYS = frozenset([
     "music_keep_screen_on",
 ])
 
+# `default_tab:<group_id>`, which is what the SPA writes (group-page.js). The
+# suffix used to be unchecked, and the route is `{key:path}`, so any string of
+# any length was a distinct key: one account could write unbounded rows into a
+# table shared with everyone, each with an unbounded `value` (the column is
+# Text). A key over 64 characters was also not a 400 but a 500 — the column is
+# String(64), which PostgreSQL enforces and SQLite does not, so it would have
+# appeared in production and in no test.
+_PREF_GROUP_KEY = re.compile(
+    r"^default_tab:[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}"
+    r"-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$")
+
+# A person cannot be in more groups than this on one hub without noticing; the
+# bound is on rows, because that is what is shared.
+MAX_PREFERENCES_PER_ACCOUNT = 500
+MAX_PREFERENCE_VALUE = 256
+
+
 def _valid_pref_key(key: str) -> bool:
     if key in ALLOWED_PREF_KEYS:
         return True
-    if key.startswith("default_tab:"):
-        return True
-    return False
+    return bool(_PREF_GROUP_KEY.match(key))
 
 
 @router.get("/me/preferences")
@@ -1005,7 +1075,9 @@ async def set_preference(
     db: AsyncSession = Depends(get_db),
 ):
     if not _valid_pref_key(key):
-        raise HTTPException(status_code=400, detail=f"Unknown preference key: {key}")
+        raise HTTPException(status_code=400, detail=f"Unknown preference key: {key[:80]}")
+    if len(body.value) > MAX_PREFERENCE_VALUE:
+        raise HTTPException(status_code=422, detail="Preference value too long")
     result = await db.execute(
         select(UserPreference).where(
             UserPreference.user_id == current_user.id,
@@ -1014,6 +1086,12 @@ async def set_preference(
     if pref:
         pref.value = body.value
     else:
+        held = (await db.execute(
+            select(func.count()).select_from(UserPreference)
+            .where(UserPreference.user_id == current_user.id))).scalar() or 0
+        if held >= MAX_PREFERENCES_PER_ACCOUNT:
+            raise HTTPException(
+                status_code=429, detail="Too many stored preferences")
         db.add(UserPreference(
             user_id=current_user.id, key=key, value=body.value))
     await db.commit()