siftlode/backend/tests/test_auth_session.py

"""Auth / session security core (TEST-1), DB-lane.

The wallet impersonation gate, epoch-based session revocation, single-use tokens and the
enumeration-safe uniform 401 are the security mag, and had zero tests (only `_oauth_redirect_uri` was
covered). Starlette's SessionMiddleware round-trips the signed cookie under TestClient, so the whole
browser-less flow is exercisable. Uses the real auth stack via the `client` fixture (NO current_user
override) so login, the wallet gate and epoch revoke are the code under test.

Note: `/api/me` is the bootstrap probe — it returns 200 `{authenticated: False}` when logged out, not a
401. A revoked session is therefore checked on a PROTECTED endpoint (`/api/feed`), which does 401.
"""

from datetime import timedelta

from fastapi.testclient import TestClient

from app.auth import ACTIVE_ACCOUNT_HEADER
from app.main import app

_PW = "correct horse battery"

# The `client`/`as_user` fixtures bake in the post-setup `_configured` state, so no per-test patch here.


def _login(client, email, password):
    return client.post("/auth/password-login", json={"email": email, "password": password})


def test_password_login_then_me_is_authenticated(client, make_user):
    make_user(email="alice@example.com", password=_PW)
    assert _login(client, "alice@example.com", _PW).status_code == 200
    me = client.get("/api/me")
    assert me.status_code == 200
    assert me.json()["authenticated"] is True
    assert me.json()["email"] == "alice@example.com"


def test_wrong_password_and_unknown_email_give_a_byte_identical_401(client, make_user):
    make_user(email="bob@example.com", password=_PW)
    wrong_pw = _login(client, "bob@example.com", "not the password")
    unknown = _login(client, "nobody@example.com", _PW)
    assert wrong_pw.status_code == 401 and unknown.status_code == 401
    # Identical body: a wrong password and a non-existent account must be indistinguishable (no
    # account-enumeration oracle).
    assert wrong_pw.json() == unknown.json()


def test_a_non_wallet_account_header_falls_back_to_the_default(client, make_user):
    # The X-Siftlode-Account header may only re-target an account already signed into THIS browser
    # (in the session wallet). A header naming someone else must be ignored, not honored — otherwise
    # it's an impersonation gate hole.
    make_user(email="alice2@example.com", password=_PW)
    mallory = make_user(email="mallory@example.com", password=_PW)
    assert _login(client, "alice2@example.com", _PW).status_code == 200
    me = client.get("/api/me", headers={ACTIVE_ACCOUNT_HEADER: str(mallory.id)})
    assert me.status_code == 200
    assert me.json()["email"] == "alice2@example.com"  # still Alice, not Mallory


def test_logout_others_revokes_the_other_session_but_keeps_the_caller(client, make_user):
    # Two independent cookie jars (browsers) signed into the same account. `logout-others` from one
    # bumps the account epoch so the OTHER's cookie is rejected on its next request, while the caller
    # is re-stamped and stays in. Checked on /api/feed (a protected endpoint that 401s on revocation).
    make_user(email="carol@example.com", password=_PW)
    # A second, independent cookie jar; the `client` fixture already routed get_db to the test DB on
    # the shared app, so this instance uses it too — it just has its own session cookie.
    other = TestClient(app)
    assert _login(client, "carol@example.com", _PW).status_code == 200
    assert _login(other, "carol@example.com", _PW).status_code == 200
    assert other.get("/api/feed").status_code == 200  # both valid to start

    assert client.post("/auth/logout-others").status_code == 200
    assert other.get("/api/feed").status_code == 401  # its epoch is now stale
    assert client.get("/api/feed").status_code == 200  # the caller stays signed in


def test_reset_confirm_is_single_use_and_kills_existing_sessions(client, make_user, db):
    from app.routes.auth import _issue_token

    user = make_user(email="dave@example.com", password=_PW)
    assert _login(client, "dave@example.com", _PW).status_code == 200
    assert client.get("/api/feed").status_code == 200  # session valid at the pre-reset epoch

    raw = _issue_token(db, user, "reset", timedelta(hours=1))
    db.commit()
    ok = client.post(
        "/auth/password-reset/confirm", json={"token": raw, "password": "brand new pw!"}
    )
    assert ok.status_code == 200

    # A reset bumps the epoch (compromise response) -> the pre-reset cookie is now rejected.
    assert client.get("/api/feed").status_code == 401
    # The token is single-use: a replay is a clean 400, not a second reset.
    replay = client.post(
        "/auth/password-reset/confirm", json={"token": raw, "password": "another pw!!"}
    )
    assert replay.status_code == 400


def test_activating_an_unverified_account_drops_its_preset_password(db, make_user):
    # SEC-10 (account pre-emption): an attacker registers victim@x with a password before the approved
    # owner arrives. The address was never verified, so admin approval must NOT hand that pre-set
    # password a working login — it is dropped and any session it minted is revoked.
    from app.routes.admin import _activate_user_by_email

    victim = make_user(
        email="victim@example.com",
        password="attacker-chosen",
        email_verified=False,
        is_active=False,
    )
    epoch0 = victim.session_epoch or 0
    _activate_user_by_email(db, "victim@example.com")
    assert victim.is_active is True
    assert victim.password_hash is None
    assert (victim.session_epoch or 0) == epoch0 + 1


def test_activating_a_verified_account_keeps_its_password(db, make_user):
    # A legit account whose email IS verified (e.g. a no-SMTP registration) keeps its password on
    # approval — the guard must not break the normal flow.
    from app.routes.admin import _activate_user_by_email

    user = make_user(
        email="legit@example.com", password="my own password", email_verified=True, is_active=False
    )
    _activate_user_by_email(db, "legit@example.com")
    assert user.is_active is True
    assert user.password_hash is not None


def _register_and_fetch(db, email: str):
    """Run the /register background worker, then read the users/invites rows back through a FRESH
    query. The worker owns its own session, so anything it failed to commit is simply not there."""
    from sqlalchemy import select as _select

    from app.models import Invite, User
    from app.routes.auth import _register_account

    _register_account(email, _PW)
    db.expire_all()
    user = db.execute(_select(User).where(User.email == email)).scalar_one_or_none()
    invite = db.execute(_select(Invite).where(Invite.email == email)).scalar_one_or_none()
    return user, invite


def test_registration_of_a_pre_approved_email_persists_the_account_without_smtp(db, monkeypatch):
    # THE regression (the only one of these that fails without the explicit commit in
    # _register_account): an admin pre-approves an address, the person then registers, and the
    # account silently never existed. upsert_pending_invite returns early WITHOUT committing for an
    # already-approved email, and the only other commit on this path (_issue_token) is gated on
    # SMTP — so on a no-SMTP instance the flushed user was rolled back at session close.
    from app.models import Invite
    from app.routes import auth as auth_mod

    monkeypatch.setattr(auth_mod.email_mod, "email_enabled", lambda: False)
    db.add(Invite(email="roland@example.com", status="approved", decided_by="admin@example.com"))
    db.commit()

    user, invite = _register_and_fetch(db, "roland@example.com")

    assert user is not None, "pre-approved registration was rolled back — the account is gone"
    assert user.password_hash is not None
    assert user.email_verified is True  # no SMTP => ownership can't be proven by link
    assert invite.status == "approved"  # an existing approval is never disturbed
    # With no SMTP there is no attestation to wait for, so the admin's standing approval IS the gate
    # and the account is usable at once. Otherwise it would sit inactive behind an invite that is no
    # longer pending — invisible in the Access-requests queue, and approve_invite 409s on it.
    assert user.is_active is True


def test_registration_of_an_unknown_email_still_persists_account_and_pending_invite(
    db, monkeypatch
):
    # Guard, not a regression test: this branch always worked (upsert INSERTs the invite and commits
    # the same transaction). It must keep working — and the account must stay inactive, since nobody
    # has approved this address.
    from app.routes import auth as auth_mod

    monkeypatch.setattr(auth_mod.email_mod, "email_enabled", lambda: False)

    user, invite = _register_and_fetch(db, "stranger@example.com")

    assert user is not None
    assert user.is_active is False
    assert invite is not None and invite.status == "pending"


def test_registration_of_a_previously_denied_email_revives_the_request(db, monkeypatch):
    # Guard: a denied person may ask again — upsert_pending_invite flips the invite back to pending
    # (and commits) so the admin sees a fresh request. The stale decision must not leak into the new
    # account's state, and re-registering must not be a dead end.
    from app.models import Invite
    from app.routes import auth as auth_mod

    monkeypatch.setattr(auth_mod.email_mod, "email_enabled", lambda: False)
    db.add(Invite(email="denis@example.com", status="denied", decided_by="admin@example.com"))
    db.commit()

    user, invite = _register_and_fetch(db, "denis@example.com")

    assert user is not None
    assert user.is_active is False
    assert invite.status == "pending"
    assert invite.decided_at is None and invite.decided_by is None


def test_registration_of_a_pre_approved_email_persists_the_account_with_smtp(db, monkeypatch):
    # Guard, not a regression test: with SMTP on, _issue_token's own commit already persisted the
    # user, so this branch was never broken and must stay unbroken. Here the account waits for the
    # verify link (see the activation test below), so it is inactive and unverified.
    from app.models import Invite
    from app.routes import auth as auth_mod

    sent: list[str] = []
    monkeypatch.setattr(auth_mod.email_mod, "email_enabled", lambda: True)
    monkeypatch.setattr(auth_mod.email_mod, "send_verify_email", lambda to, link: sent.append(to))
    monkeypatch.setattr(auth_mod.email_mod, "send_admin_new_request", lambda admins, to: True)
    db.add(Invite(email="rita@example.com", status="approved", decided_by="admin@example.com"))
    db.commit()

    user, _ = _register_and_fetch(db, "rita@example.com")

    assert user is not None
    assert user.email_verified is False  # must confirm the address
    assert user.is_active is False
    assert sent == ["rita@example.com"]


def test_verifying_an_email_activates_a_pre_approved_account(client, db, make_user):
    # Confirming the emailed link attests ownership, so an approval the admin granted before the
    # person signed up is honoured right there. Without this the SMTP-on pre-approved registrant is
    # stuck: their invite is already decided, so it never reaches the Access-requests queue and
    # approve_invite 409s on it.
    from app.models import Invite
    from app.routes.auth import _issue_token

    user = make_user(email="rita@example.com", password=_PW, email_verified=False, is_active=False)
    db.add(Invite(email="rita@example.com", status="approved", decided_by="admin@example.com"))
    db.commit()
    raw = _issue_token(db, user, "verify", timedelta(hours=1))

    assert client.post("/auth/verify", json={"token": raw}).status_code == 200

    db.expire_all()
    assert user.email_verified is True
    assert user.is_active is True
    assert _login(client, "rita@example.com", _PW).status_code == 200


def test_verifying_an_email_does_not_activate_an_unapproved_account(client, db, make_user):
    # The mirror guard: verification proves the address, it does not grant access. With no approved
    # invite the account stays inactive and sign-in is still refused.
    from app.routes.auth import _issue_token

    user = make_user(
        email="nobody@example.com", password=_PW, email_verified=False, is_active=False
    )
    raw = _issue_token(db, user, "verify", timedelta(hours=1))

    assert client.post("/auth/verify", json={"token": raw}).status_code == 200

    db.expire_all()
    assert user.email_verified is True
    assert user.is_active is False
    assert _login(client, "nobody@example.com", _PW).status_code == 403


def test_register_route_creates_the_account_end_to_end(client, db, monkeypatch):
    # The tests above call _register_account directly, so nothing covered the ROUTE: its
    # allow_registration gate, its rate limiter and the background.add_task wiring that makes the
    # whole flow reachable at all. TestClient runs BackgroundTasks after the response, so a single
    # POST exercises the real path end to end.
    from sqlalchemy import select as _select

    from app.models import User
    from app.routes import auth as auth_mod

    monkeypatch.setattr(auth_mod.email_mod, "email_enabled", lambda: False)

    r = client.post("/auth/register", json={"email": "newbie@example.com", "password": _PW})

    assert r.status_code == 200 and r.json() == {"status": "ok"}
    db.expire_all()
    user = db.execute(_select(User).where(User.email == "newbie@example.com")).scalar_one_or_none()
    assert user is not None, "POST /auth/register answered ok but created nothing"
    assert user.password_hash is not None


def test_register_route_is_refused_when_registration_is_closed(client, db, monkeypatch):
    # The gate is an explicit 403 with nothing created — not a uniform "ok" that silently drops it.
    from sqlalchemy import select as _select

    from app import sysconfig
    from app.models import User
    from app.routes import auth as auth_mod

    monkeypatch.setattr(auth_mod.email_mod, "email_enabled", lambda: False)
    sysconfig.set_value(db, "allow_registration", False)

    r = client.post("/auth/register", json={"email": "closed@example.com", "password": _PW})

    assert r.status_code == 403
    db.expire_all()
    assert (
        db.execute(_select(User).where(User.email == "closed@example.com")).scalar_one_or_none()
        is None
    )