siftlode/backend/tests/test_google_callback.py

"""The Google callback's account-matching branches, through the real route, DB-lane.

Each `select()` here was rewritten from a `.query()` without a test that drove it, and one branch —
adopting an existing row — is where a no-SMTP squat of an env-allowed address turned into an admin
account with the squatter's password. These pin the adoption (and what it revokes), its two refusals,
the refused sign-in that doubles as an access request, and the link flow's conflict answers.
"""

import pytest
from sqlalchemy import select
from starlette.responses import RedirectResponse

from app.config import settings
from app.models import Invite, User
from app.routes import auth as auth_routes

_PW = "correct horse battery"


class _FakeGoogle:
    """authlib's client, reduced to the two calls the routes make: the redirect out to Google and
    the token exchange when Google redirects back."""

    def __init__(self, userinfo: dict):
        self.userinfo = userinfo

    async def authorize_redirect(self, _request, _redirect_uri, **_kw):
        return RedirectResponse(url="/fake-google")

    async def authorize_access_token(self, _request):
        return {"userinfo": self.userinfo, "access_token": "at", "scope": "openid email profile"}


@pytest.fixture
def google(monkeypatch):
    from cryptography.fernet import Fernet

    from app import security

    # The callback stores the OAuth grant encrypted; a throwaway key stands in for TOKEN_ENCRYPTION_KEY.
    monkeypatch.setattr(security, "_fernet", Fernet(Fernet.generate_key()))
    fake = _FakeGoogle({})
    monkeypatch.setattr(auth_routes, "google_oauth", lambda _db: fake)
    return fake


def _sign_in(client, google, **userinfo):
    google.userinfo = {"email_verified": True, **userinfo}
    return client.get("/auth/callback", follow_redirects=False)


def _rows(db, model, column, value):
    db.expire_all()
    return db.execute(select(model).where(column == value)).scalars().all()


def test_a_no_smtp_squat_of_an_admin_address_loses_its_password(client, db, google, monkeypatch):
    # A stranger registers the ADMIN_EMAILS address (a sibling spelling will do) before its owner
    # does. Without SMTP the row is marked verified, and an env-allowed address gets no invite row
    # to gate it — so the owner's first Google sign-in activates THIS row as admin. The squatter's
    # password must not survive that.
    monkeypatch.setattr(auth_routes.email_mod, "email_enabled", lambda db=None: False)
    monkeypatch.setattr(auth_routes, "notify_admins_of_request", lambda *_a, **_k: None)
    monkeypatch.setattr(settings, "admin_emails", "bob@gmail.com")
    auth_routes._register_account("b.ob@gmail.com", "the squatter's own")
    [squat] = _rows(db, User, User.email, "b.ob@gmail.com")
    assert squat.email_verified is True and squat.is_active is False

    r = _sign_in(client, google, sub="sub-bob", email="bob@gmail.com")

    assert r.headers["location"] == "/"
    [owner] = _rows(db, User, User.email_canonical, squat.email_canonical)
    assert (owner.id, owner.role, owner.is_active) == (squat.id, "admin", True)
    assert owner.password_hash is None and owner.password_revoked_at is not None
    assert (
        client.post(
            "/auth/password-login",
            json={"email": "b.ob@gmail.com", "password": "the squatter's own"},
        ).status_code
        == 401
    )


def test_an_activated_squat_loses_its_password_when_the_sign_in_makes_it_admin(
    client, db, google, make_user, monkeypatch
):
    # The same squat after an admin clicked Activate on it: activation vouches for access, not for
    # who chose the password, and without SMTP nothing else ever proved the mailbox. ADMIN_EMAILS
    # must not raise the squatter's live session to admin.
    monkeypatch.setattr(auth_routes.email_mod, "email_enabled", lambda db=None: False)
    monkeypatch.setattr(settings, "admin_emails", "bob@gmail.com")
    squat = make_user(email="b.ob@gmail.com", password="the squatter's own", is_active=True)
    squatter = client.__class__(client.app)
    assert (
        squatter.post(
            "/auth/password-login",
            json={"email": "b.ob@gmail.com", "password": "the squatter's own"},
        ).status_code
        == 200
    )

    _sign_in(client, google, sub="sub-bob", email="bob@gmail.com")

    db.refresh(squat)
    assert squat.role == "admin"
    assert squat.password_hash is None and squat.password_revoked_at is not None
    assert squatter.get("/api/me").json() == {"authenticated": False}


def test_with_smtp_too_an_account_made_admin_loses_its_password(
    client, db, google, make_user, monkeypatch
):
    # SMTP configured now says nothing about a row verified before it was: the rule does not ask.
    monkeypatch.setattr(auth_routes.email_mod, "email_enabled", lambda db=None: True)
    monkeypatch.setattr(settings, "admin_emails", "own@example.com")
    user = make_user(email="own@example.com", password=_PW, is_active=True)

    _sign_in(client, google, sub="sub-own", email="own@example.com")

    db.refresh(user)
    assert user.role == "admin" and user.password_hash is None
    # …and the owner, signed in through Google, is not thrown out by the epoch bump.
    assert client.get("/api/me").json()["id"] == user.id


def test_a_later_sign_in_that_makes_the_account_admin_drops_its_password(
    client, db, google, make_user, monkeypatch
):
    # Adopted while the address was only allowed (the password kept: nothing promoted it), then
    # added to ADMIN_EMAILS: the promotion happens on a sign-in matched by Google identity.
    user = make_user(email="later@example.com", password=_PW, is_active=True, google_sub="sub-l")
    monkeypatch.setattr(settings, "admin_emails", "later@example.com")

    _sign_in(client, google, sub="sub-l", email="later@example.com")

    db.refresh(user)
    assert user.role == "admin"
    assert user.password_hash is None and user.password_revoked_at is not None


def test_an_admin_sign_in_keeps_an_admin_password(client, db, google, make_user, monkeypatch):
    # Only the promotion drops it: an account that is already admin signs in as before.
    user = make_user(email="boss@example.com", password=_PW, role="admin", google_sub="sub-b")
    monkeypatch.setattr(settings, "admin_emails", "boss@example.com")

    _sign_in(client, google, sub="sub-b", email="boss@example.com")

    db.refresh(user)
    assert user.password_hash is not None


def test_adopting_an_account_already_let_in_keeps_its_password(
    client, db, google, make_user, monkeypatch
):
    # The revoke is for passwords nobody vouched for; an active, verified account that this sign-in
    # does not make admin keeps its own.
    monkeypatch.setattr(auth_routes.email_mod, "email_enabled", lambda db=None: False)
    user = make_user(email="kept@example.com", password=_PW, is_active=True)
    monkeypatch.setattr(settings, "allowed_emails", "kept@example.com")

    _sign_in(client, google, sub="sub-kept", email="kept@example.com")

    db.refresh(user)
    assert user.google_sub == "sub-kept"
    assert user.password_hash is not None and user.password_revoked_at is None


def test_adoption_matches_the_mailbox_not_the_spelling(client, db, google, make_user, monkeypatch):
    user = make_user(email="pair.one@gmail.com", is_active=True)
    monkeypatch.setattr(settings, "admin_emails", "pairone@gmail.com")

    _sign_in(client, google, sub="sub-pair", email="pairone@gmail.com")

    assert [u.id for u in _rows(db, User, User.email_canonical, user.email_canonical)] == [user.id]
    db.refresh(user)
    assert user.google_sub == "sub-pair"


def test_a_different_google_identity_on_the_address_is_refused(
    client, db, google, make_user, monkeypatch
):
    user = make_user(email="held@example.com", google_sub="sub-first", is_active=True)
    monkeypatch.setattr(settings, "admin_emails", "held@example.com")

    r = _sign_in(client, google, sub="sub-second", email="held@example.com")

    assert "access=denied" in r.headers["location"]
    db.refresh(user)
    assert user.google_sub == "sub-first"


def test_an_unverified_google_address_does_not_adopt(client, db, google, make_user, monkeypatch):
    user = make_user(email="pending@example.com", is_active=False, email_verified=False)
    monkeypatch.setattr(settings, "admin_emails", "pending@example.com")

    google.userinfo = {"sub": "sub-unv", "email": "pending@example.com", "email_verified": False}
    r = client.get("/auth/callback", follow_redirects=False)

    assert "access=denied" in r.headers["location"]
    db.refresh(user)
    assert (user.google_sub, user.is_active) == (None, False)


def test_a_refused_sign_in_records_a_request_in_the_reported_language(
    client, db, google, monkeypatch
):
    notified: list[str] = []
    monkeypatch.setattr(
        auth_routes, "notify_admins_of_request", lambda _db, _bg, email: notified.append(email)
    )

    r = _sign_in(client, google, sub="sub-new", email="new@example.com", locale="hu-HU")

    assert "access=requested" in r.headers["location"]
    [invite] = _rows(db, Invite, Invite.email, "new@example.com")
    assert (invite.status, invite.lang) == ("pending", "hu")
    assert notified == ["new@example.com"]
    assert _rows(db, User, User.email, "new@example.com") == []


def _start_link(client, email: str):
    assert (
        client.post("/auth/password-login", json={"email": email, "password": _PW}).status_code
        == 200
    )
    assert client.get("/auth/link", follow_redirects=False).headers["location"] == "/fake-google"


def test_linking_an_identity_another_account_holds_is_a_conflict(client, db, google, make_user):
    me = make_user(email="me@example.com")
    make_user(email="them@example.com", google_sub="sub-theirs")
    _start_link(client, "me@example.com")

    r = _sign_in(client, google, sub="sub-theirs", email="them@example.com")

    assert r.headers["location"] == "/?link=conflict"
    db.refresh(me)
    assert me.google_sub is None


def test_linking_a_second_identity_is_a_mismatch(client, db, google, make_user):
    me = make_user(email="me2@example.com", google_sub="sub-mine")
    _start_link(client, "me2@example.com")

    r = _sign_in(client, google, sub="sub-other", email="other@example.com")

    assert r.headers["location"] == "/?link=mismatch"
    db.refresh(me)
    assert me.google_sub == "sub-mine"


def test_linking_a_free_identity_attaches_it(client, db, google, make_user):
    me = make_user(email="me3@example.com")
    _start_link(client, "me3@example.com")

    r = _sign_in(client, google, sub="sub-free", email="me3.google@example.com")

    assert r.headers["location"] == "/?link=ok"
    db.refresh(me)
    assert (me.google_sub, me.google_email) == ("sub-free", "me3.google@example.com")