""" The hub can send mail from its own domain, to anywhere, with no authentication to its local Postfix. That is an open relay unless something bounds it, and what bounds it has to be one place, or the next `send_*` helper added will not have it. Three API paths reach that ability and two of them mail an address the caller types: POST /v1/users/register an address nobody has verified PATCH /v1/users/me an address nobody has verified, signed in POST /v1/users/password/reset only the address already on file POST /v1/groups/{id}/invite-notify only a registered member's address The widest was the register *resend* branch: no token, no captcha, and the username and address are the caller's own from a moment ago, so registering a victim's address once bought the right to mail them at the endpoint's rate limit indefinitely. The bound that matters is per **recipient**, because that is what a person being mail-bombed experiences, and no combination of accounts, addresses or endpoints moves it. Per-account and per-IP limits sit above it and bound a caller, which is a different and weaker thing. """ import base64 import pytest from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey from meshbay_common.crypto import pk_to_b64 from meshbay_hub import mail as mail_mod from meshbay_hub.api import users as users_mod # Captured at import, before conftest's autouse fixture replaces them for every # other test in the suite: one stubs `_send` out entirely and the other skips # the registration mail. This file is about the code they stub. _REAL_SEND = mail_mod._send _REAL_CREATE_AND_SEND = users_mod._create_and_send_verification @pytest.fixture(autouse=True) def _real_mail_path(monkeypatch): """Undo conftest's stubs, for the tests that go through the API. The direct tests below call `_REAL_SEND` rather than `mail_mod._send`: depending on a fixture to *un*-patch a module global is a coupling that passes alone and fails in a full run, which is how this file first behaved. """ monkeypatch.setattr(mail_mod, "_send", _REAL_SEND) monkeypatch.setattr(users_mod, "_create_and_send_verification", _REAL_CREATE_AND_SEND) assert mail_mod._send is _REAL_SEND @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) @pytest.fixture def wire(monkeypatch): """Everything that actually reaches Postfix, captured at the socket.""" sent = [] class _FakeSMTP: def __init__(self, *a, **kw): pass def __enter__(self): return self def __exit__(self, *a): return False def send_message(self, msg): sent.append((msg["To"], msg["Subject"])) monkeypatch.setattr(mail_mod.smtplib, "SMTP", _FakeSMTP) return sent def _msg(to: str): from email.message import EmailMessage m = EmailMessage() m["From"] = "noreply@test" m["To"] = to m["Subject"] = "test" m.set_content("test") return m # ── The gate itself ────────────────────────────────────────────────────────── def test_a_send_must_name_a_purpose(wire): """Keyword-required, so a helper that forgets is a TypeError rather than an unrestricted send.""" with pytest.raises(TypeError): _REAL_SEND(_msg("someone@example.test")) def test_only_the_four_purposes_send(wire): assert _REAL_SEND(_msg("a@example.test"), purpose="registration") assert not _REAL_SEND(_msg("b@example.test"), purpose="marketing") assert not _REAL_SEND(_msg("c@example.test"), purpose="") 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.""" victim = "victim@example.test" assert _REAL_SEND(_msg(victim), purpose="registration") assert not _REAL_SEND(_msg(victim), purpose="password_reset"), ( "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") 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 def test_the_same_recipient_is_one_recipient_however_it_is_written(wire): """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 def test_the_instance_has_a_ceiling_no_account_can_buy_past(wire): """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 def test_a_refusal_never_names_the_address(): """This line goes to the journal, which is not where addresses live. Listens on the module's own logger rather than through `caplog`: the hub configures logging when an app is built, so whether pytest's capture sees anything depends on which tests ran first. In the full suite it saw nothing, and the assertion that had to hold — no address in the line — was never reached. """ import logging records: list[str] = [] class _Capture(logging.Handler): 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 # the full suite and everything on its own. handler = _Capture() previous_level, previously_disabled = mail_mod.log.level, mail_mod.log.disabled mail_mod.log.addHandler(handler) mail_mod.log.setLevel(logging.WARNING) # `.disabled` as well as the level and the manager-wide switch. Something # earlier in a full run leaves this logger disabled — pytest's own capture # plugin toggles loggers between tests — and a disabled logger drops the # record before any handler sees it. Alone, none of this is needed; the # first two versions of this test passed alone and failed in the suite for # two different reasons, which is the whole argument for not reading a # global's state in a test without pinning it first. mail_mod.log.disabled = False logging.disable(logging.NOTSET) try: _REAL_SEND(_msg("private@example.test"), purpose="marketing") finally: mail_mod.log.removeHandler(handler) mail_mod.log.setLevel(previous_level) mail_mod.log.disabled = previously_disabled assert records, "a refusal must leave a trace" assert not any("private@example.test" in line for line in records) assert any("marketing" in line for line in records), ( "the trace has to say what was refused to be worth writing") # ── The doors, driven through the API ──────────────────────────────────────── async def _register(client, username: str, email: str, captcha=None): sk_ed, sk_x = Ed25519PrivateKey.generate(), X25519PrivateKey.generate() return await client.post("/v1/users/register", json={ "username": username, "email": email, "auth_key": base64.b64encode(b"k" * 32).decode(), "pk_user_ed25519": pk_to_b64(sk_ed.public_key()), "pk_user_x25519": pk_to_b64(sk_x.public_key()), }) @pytest.mark.asyncio async def test_the_register_resend_branch_cannot_be_hammered(client, wire): """ No token and no captcha reach this branch. Registering a victim's address once used to buy the right to mail them at the endpoint's rate limit for as long as the account stayed pending. """ victim = "bombing-target@example.test" r = await _register(client, "relay_a", victim) assert r.status_code == 201, r.text assert len(wire) == 1 for _ in range(5): r = await _register(client, "relay_a", victim) assert r.status_code == 201, r.text assert len(wire) == 1, f"{len(wire)} mails to one address from one sign-up" async def _signed_in(client, db_session, username: str, email: str) -> dict: """Register, activate, sign in. Activated by hand because this file restores the real verification mail, which conftest stubs out for the rest of the suite — so the account is genuinely `pending` here, exactly as it would be in production, and a pending account cannot log in. """ from sqlalchemy import select, update from meshbay_hub.db.models import User r = await _register(client, username, email) assert r.status_code == 201, r.text await db_session.execute( update(User).where(User.username == username).values(status="active")) await db_session.commit() assert (await db_session.execute( select(User.status).where(User.username == username))).scalar() == "active" r = await client.post("/v1/users/login", json={ "username": username, "auth_key": base64.b64encode(b"k" * 32).decode()}) assert r.status_code == 200, r.text return {"Authorization": f"Bearer {r.json()['access_token']}"} @pytest.mark.asyncio async def test_a_signed_in_account_cannot_mail_a_stranger_repeatedly( client, db_session, wire, monkeypatch): """ `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. """ 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 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 @pytest.mark.asyncio async def test_the_daily_ceiling_survives_the_row_being_deleted( client, db_session, wire, monkeypatch): """ 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. """ 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-one-more@example.test"}) assert r.status_code == 429, ( "the ceiling was counted from a table the handler empties")