siftlode/backend/tests/test_review_round.py

"""E9.5 review round — the invariants the epic's own review found unpinned.

Every test here corresponds to a defect a reviewer could still introduce with the whole suite green:
a mail whose body never renders in any test, an activation that only works for the exact spelling, a
"we already wrote to them" stamp for a mail that was dropped, an env allow-list that answers a
different question than the gate it backs, and the address pre-emption that E9.5 S1's argument
covered only while the address stayed unverified.
"""

from datetime import datetime, timedelta, timezone

import pytest
from sqlalchemy import select

from app import auth as auth_core
from app import email as email_mod
from app import mailtext
from app.config import settings
from app.models import AuthToken, Invite, User
from app.routes import auth as auth_routes

_PW = "correct horse battery"


@pytest.fixture
def wire(monkeypatch):
    """Capture what `_send` would have put on the wire. Mails are the one surface where a body that
    never renders in a test is a body that first renders in someone's inbox."""
    out: list[tuple[list[str], str, str]] = []
    monkeypatch.setattr(
        email_mod,
        "_send",
        lambda to, subject, body, reply_to=None: out.append((to, subject, body)) is None,
    )
    return out


@pytest.fixture
def smtp_on(monkeypatch):
    monkeypatch.setattr(email_mod, "email_enabled", lambda db=None: True)


# --- every mail actually renders ------------------------------------------------------------------

# One call per `send_*`, with arguments of the right shape. The parity test compares HU against EN
# and can therefore never catch a template whose placeholders disagree with its CALLER — that
# mismatch is a KeyError inside a background task, i.e. a mail that silently never goes out. Three
# mailers this epic rewrote (`password_reset`, `role_changed`, `account_unsuspended`) had no test
# that executed their `.format()` at all.
_EVERY_MAIL = [
    ("send_access_approved", ("to@example.com", "LANG", "https://x/#verify=t", True, 168)),
    ("send_access_approved", ("to@example.com", "LANG", None, True, 24)),
    ("send_access_approved", ("to@example.com", "LANG", None, False, 24)),
    ("send_access_invitation", ("to@example.com", "LANG", "https://x/#verify=t", True, 168)),
    ("send_access_invitation", ("to@example.com", "LANG", None, False, 24)),
    ("send_access_denied", ("to@example.com", "LANG")),
    ("send_role_changed", ("to@example.com", "LANG", "admin", "op@example.com")),
    ("send_role_changed", ("to@example.com", "LANG", "user", None)),
    ("send_account_unsuspended", ("to@example.com", "LANG", "op@example.com")),
    ("send_account_deleted", ("to@example.com", "LANG", None)),
    ("send_account_suspended", ("to@example.com", "LANG", "op@example.com")),
    ("send_account_suspended", ("to@example.com", "LANG", None)),
    ("send_admin_new_request", (["a@example.com"], "LANG", "who@example.com")),
    ("send_verify_email", ("to@example.com", "LANG", "https://x/#verify=t", 24)),
    ("send_password_reset", ("to@example.com", "LANG", "https://x/#reset=t")),
    ("send_test", ("to@example.com", "LANG")),
]


@pytest.mark.parametrize("lang", mailtext.SUPPORTED)
@pytest.mark.parametrize("name,args", _EVERY_MAIL, ids=[f"{n}{a[2:3]}" for n, a in _EVERY_MAIL])
def test_every_mail_renders_in_every_language(wire, name, args, lang):
    getattr(email_mod, name)(*[lang if a == "LANG" else a for a in args])

    (_, subject, body) = wire[0]
    assert subject and body
    # A template that lost its interpolation renders the brace text verbatim.
    assert "{" not in subject and "{" not in body


def test_a_weeks_link_is_described_in_days_not_hours(wire):
    # The S6 approval token lives seven days, and the mail used to say "expires in 168 hours".
    email_mod.send_access_approved("to@example.com", "en", "https://x/#verify=t", True, 168)
    assert "7 days" in wire[0][2] and "168" not in wire[0][2]

    email_mod.send_verify_email("to@example.com", "hu", "https://x/#verify=t", 24)
    assert "24 óra" in wire[1][2]


# --- what a second registration may and may not do -------------------------------------------------


def test_a_later_registration_never_touches_an_existing_password(db, client, smtp_on, wire):
    """The rule E9.5's review round 2 put back, having tried the opposite and found it worse.

    "While the address is unverified, the newest registration owns it" looked like a fix for address
    pre-emption. It is symmetric — an attacker can always be last — so it closed nothing, and it
    handed an unauthenticated caller the power to overwrite a real registrant's chosen password and
    delete the confirmation link already sitting in their inbox. On a SEC-10-stripped row it would
    have installed a password where the whole point is that there is none. Reverted; the owner's
    recourse is password-reset, which proves the mailbox. Residual logged as SEC-16.
    """
    client.post(
        "/auth/register", json={"email": "victim@example.com", "password": "the first password"}
    )
    db.expire_all()
    user = db.execute(select(User).where(User.email == "victim@example.com")).scalar_one()
    before = user.password_hash

    client.post(
        "/auth/register", json={"email": "victim@example.com", "password": "the second password"}
    )

    db.expire_all()
    assert db.get(User, user.id).password_hash == before


def test_a_later_registration_leaves_a_link_already_in_flight_alive(db, client, smtp_on, wire):
    client.post(
        "/auth/register", json={"email": "victim@example.com", "password": "the first password"}
    )
    db.expire_all()
    user = db.execute(select(User).where(User.email == "victim@example.com")).scalar_one()
    first = db.execute(select(AuthToken.id).where(AuthToken.user_id == user.id)).scalars().all()

    client.post(
        "/auth/register", json={"email": "victim@example.com", "password": "the second password"}
    )

    db.expire_all()
    live = db.execute(select(AuthToken.id).where(AuthToken.user_id == user.id)).scalars().all()
    # The re-send adds a link; it must not invalidate the one the owner may already be holding.
    assert set(first) <= set(live)


def test_a_stripped_account_keeps_its_recovery_stamp(db, client, smtp_on, make_user):
    """Prod carries accounts SEC-10 left with no password and a `password_revoked_at` stamp. That
    stamp is the only reason password-reset still accepts them; an anonymous POST must not be able to
    clear it, nor to put a password on a row whose whole state is "there is none"."""
    stripped = make_user(
        email="stripped@example.com",
        password=None,
        email_verified=False,
        is_active=True,
        password_revoked_at=datetime.now(timezone.utc),
    )

    client.post(
        "/auth/register", json={"email": "stripped@example.com", "password": "an attacker password"}
    )

    db.expire_all()
    row = db.get(User, stripped.id)
    assert row.password_hash is None and row.password_revoked_at is not None


def test_a_denied_sibling_is_not_reopened_while_a_pending_row_exists(db, client):
    # Canonical siblings exist on prod. Picking the denied one reopens it and re-notifies every
    # admin, putting one person in the queue twice — and the exact-spelling tie-break made that
    # deterministic rather than merely possible.
    db.add_all(
        [
            Invite(email="pair.one@gmail.com", status="pending"),
            Invite(email="pairone@gmail.com", status="denied"),
        ]
    )
    db.commit()

    client.post("/auth/request-access", json={"email": "pairone@gmail.com"})

    db.expire_all()
    rows = {i.email: i.status for i in db.execute(select(Invite)).scalars().all()}
    assert rows == {"pair.one@gmail.com": "pending", "pairone@gmail.com": "denied"}


# --- the env bootstrap answers the same question as the gate --------------------------------------


def test_the_env_allow_list_is_matched_canonically(db, monkeypatch):
    # An operator writing the dotted spelling into ALLOWED_EMAILS was refused when Google answered
    # with the undotted primary — the 2026-09-03 lockout, on the bootstrap path.
    monkeypatch.setattr(settings, "allowed_emails", "first.last@gmail.com")

    assert auth_core.is_allowed(db, "firstlast@gmail.com") is True
    assert auth_core.is_allowed(db, "first.last+tag@gmail.com") is True
    assert auth_core.is_allowed(db, "someone.else@gmail.com") is False


def test_an_env_granted_address_cannot_be_shadowed_by_a_stranger(db, client, monkeypatch):
    # `is_allowed` short-circuits on ANY invite row for the canonical form, so before this an
    # unauthenticated caller could request access under a sibling spelling and permanently shadow an
    # env grant. With the env list matched canonically the request never opens a row.
    monkeypatch.setattr(settings, "allowed_emails", "bob@gmail.com")

    client.post("/auth/request-access", json={"email": "b.ob@gmail.com"})

    assert db.execute(select(Invite)).scalars().all() == []
    assert auth_core.is_allowed(db, "bob@gmail.com") is True


def test_a_dotted_admin_emails_entry_still_bootstraps_its_admin(monkeypatch):
    monkeypatch.setattr(settings, "admin_emails", "the.operator+siftlode@gmail.com")

    assert "theoperator@gmail.com" in auth_core.env_admin_canonicals()


# --- approving decides about a MAILBOX, not a spelling --------------------------------------------


def test_approving_one_spelling_activates_the_account_registered_under_another(
    db, make_user, as_user, smtp_on, wire
):
    # The 2026-09-03 incident in one test: the admin approves the address as typed into the request,
    # and the account exists under the other Gmail spelling of the same mailbox.
    admin = make_user(email="admin@example.com", role="admin")
    user = make_user(email="dotless@gmail.com", password=_PW, is_active=False)
    user.email = "dotless@gmail.com"
    inv = Invite(email="dot.less@gmail.com", status="pending")
    db.add(inv)
    db.commit()

    r = as_user(admin).post(f"/api/admin/invites/{inv.id}/approve")

    assert r.status_code == 200
    db.expire_all()
    assert db.get(User, user.id).is_active is True


def test_deleting_one_of_two_accounts_leaves_the_other_one_allowed(db, make_user, as_user):
    # Prod carries two accounts for one Gmail mailbox. A canonical invite sweep on delete took the
    # survivor's grant with it, refusing their next sign-in — this epic's founding bug, self-inflicted.
    admin = make_user(email="admin@example.com", role="admin")
    doomed = make_user(email="pair.one@gmail.com", password=_PW)
    make_user(email="pairone@gmail.com", password=_PW)
    db.add_all([Invite(email="pair.one@gmail.com", status="approved")])
    db.commit()

    r = as_user(admin).delete(f"/api/admin/users/{doomed.id}")

    assert r.status_code == 200
    assert auth_core.is_allowed(db, "pairone@gmail.com") is True


# --- a button an admin pressed must not report a send that did not happen -------------------------


def test_a_failed_invitation_is_not_recorded_as_written(
    db, make_user, as_user, smtp_on, monkeypatch
):
    monkeypatch.setattr(email_mod, "send_access_invitation", lambda *a, **kw: False)
    admin = make_user(email="admin@example.com", role="admin")
    inv = Invite(email="stranger@example.com", status="approved")
    db.add(inv)
    db.commit()

    r = as_user(admin).post(f"/api/admin/invites/{inv.id}/invite", json={"lang": "en"})

    assert r.status_code == 422 and r.json()["detail"]["reason"] == "send_failed"
    db.expire_all()
    # The column exists to answer "did I already write to them?", and the UI dims the button and
    # warns about a second mail on the strength of it. A dropped send must not make it say yes.
    assert db.get(Invite, inv.id).invited_at is None


def test_a_failed_resend_says_so_instead_of_reporting_ok(
    db, make_user, as_user, smtp_on, monkeypatch
):
    monkeypatch.setattr(email_mod, "send_verify_email", lambda *a, **kw: False)
    admin = make_user(email="admin@example.com", role="admin")
    stuck = make_user(email="stuck@example.com", password=_PW, email_verified=False)

    r = as_user(admin).post(f"/api/admin/users/{stuck.id}/resend-verification")

    assert r.status_code == 422 and r.json()["detail"]["reason"] == "send_failed"


# --- refusals and links that are already dead ------------------------------------------------------


def test_an_expired_reset_link_is_reported_dead_without_being_spent(db, client, make_user):
    # The endpoint exists because a spent or stale link used to draw the full "set a new password"
    # form. Expiry is the reason its docstring gives, and it was the one case no test covered.
    user = make_user(email="late@example.com", password=_PW)
    raw = auth_routes._issue_token(db, user, "reset", timedelta(hours=1))
    token = db.execute(select(AuthToken).where(AuthToken.user_id == user.id)).scalar_one()
    token.expires_at = datetime.now(timezone.utc) - timedelta(minutes=1)
    db.commit()

    assert client.post("/auth/password-reset/check", json={"token": raw}).json() == {"ok": False}
    r = client.post(
        "/auth/password-reset/confirm", json={"token": raw, "password": "a new one 123"}
    )
    assert r.status_code == 400


def test_a_verify_token_cannot_pass_as_a_reset_link(db, client, make_user):
    user = make_user(email="mixed@example.com", password=_PW)
    raw = auth_routes._issue_token(db, user, "verify", timedelta(hours=1))

    assert client.post("/auth/password-reset/check", json={"token": raw}).json() == {"ok": False}


def test_a_throttled_login_says_nothing_about_the_account(db, client, make_user, monkeypatch):
    # The two refusals that answer BEFORE the password is proven must stay byte-identical for a real
    # account and an unknown one; only `invalid` was pinned.
    make_user(email="real@example.com", password=_PW)
    monkeypatch.setattr(auth_routes._login_limiter, "allow", lambda key: False)

    real = client.post("/auth/password-login", json={"email": "real@example.com", "password": _PW})
    unknown = client.post(
        "/auth/password-login", json={"email": "nobody@example.com", "password": _PW}
    )

    assert real.status_code == unknown.status_code == 429
    assert real.json() == unknown.json() == {"detail": {"code": "login", "reason": "rate_limited"}}


# --- admins hear about a request in their own language --------------------------------------------


def test_two_admins_two_languages_two_mails(db, client, make_user, smtp_on, wire):
    make_user(email="hu.admin@example.com", role="admin", preferences={"language": "hu"})
    make_user(email="en.admin@example.com", role="admin", preferences={"language": "en"})

    client.post("/auth/request-access", json={"email": "asking@example.com", "lang": "en"})

    notices = [w for w in wire if "asking@example.com" in w[1]]
    assert len(notices) == 2
    by_recipient = {w[0][0]: w[1] for w in notices}
    assert by_recipient["hu.admin@example.com"] == mailtext.MAIL_TEXT["admin_new_request.subject"][
        "hu"
    ].format(requester="asking@example.com")
    assert by_recipient["en.admin@example.com"] == mailtext.MAIL_TEXT["admin_new_request.subject"][
        "en"
    ].format(requester="asking@example.com")


def test_an_env_admin_without_an_account_is_written_to_in_the_default(db, monkeypatch):
    monkeypatch.setattr(settings, "admin_emails", "bootstrap@example.com")

    assert auth_routes.admin_notify_groups(db) == {mailtext.DEFAULT: ["bootstrap@example.com"]}