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: 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_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's own password
    # is not one of them.
    user = make_user(email="kept@example.com", password=_PW, is_active=True)
    monkeypatch.setattr(settings, "admin_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")