diff options
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 |