path: root/packages/meshbay-hub/tests/test_mail_is_not_a_relay.py

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