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 authlib.integrations.starlette_client import OAuthError
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. Like authlib, it honours only the newest flow's
    state: starting a flow retires every earlier one."""

    def __init__(self, userinfo: dict):
        self.userinfo = userinfo
        self.asked: dict = {}

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

    async def authorize_access_token(self, request):
        if request.query_params.get("state") != self.asked.get("state"):
            raise OAuthError(error="mismatching_state")
        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, state: str | None = None, **userinfo):
    """Google redirecting back: by default with the state of the flow started last."""
    google.userinfo = {"email_verified": True, **userinfo}
    state = state or google.asked.get("state")
    return client.get(
        "/auth/callback", params={"state": state} if state else {}, 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 _password_login(client, email: str):
    assert (
        client.post("/auth/password-login", json={"email": email, "password": _PW}).status_code
        == 200
    )


def _connect(client, access: str):
    return client.get(f"/auth/google/connect?access={access}", follow_redirects=False)


def _start_link(client, email: str):
    _password_login(client, email)
    assert _connect(client, "identity").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")


def test_a_connect_abandoned_at_google_leaves_the_next_sign_in_a_sign_in(
    client, db, google, make_user, monkeypatch
):
    # A connect left at Google's screen (Back, a closed tab) never reaches the callback that pops
    # its marker. "Sign in as someone else" afterwards must sign that person in, not link their
    # Google identity onto the account still signed in here.
    me = make_user(email="me4@example.com")
    make_user(email="them4@example.com", google_sub="sub-them4", password=None)
    monkeypatch.setattr(settings, "allowed_emails", "them4@example.com")
    _start_link(client, "me4@example.com")
    assert client.get("/auth/login", follow_redirects=False).headers["location"] == "/fake-google"

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

    assert r.headers["location"] == "/"
    assert client.get("/api/me").json()["email"] == "them4@example.com"
    db.refresh(me)
    assert me.google_sub is None


def test_a_link_whose_session_was_revoked_meanwhile_does_not_finish(client, db, google, make_user):
    # "Log out everywhere" (or a password change) between the start and the callback: the cookie
    # still names the account, but its session no longer holds, so it may not attach an identity.
    me = make_user(email="me5@example.com")
    _start_link(client, "me5@example.com")
    me.session_epoch = (me.session_epoch or 0) + 1
    db.commit()

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

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


def test_an_older_flow_coming_back_leaves_the_connect_in_flight_a_link(
    client, db, google, make_user
):
    # Two round-trips from one browser: an older one (another tab's, or Back to Google's screen)
    # returns after a connect started. Its state is refused, and it must not spend the connect's
    # marker: the connect itself, coming back, would then land as a plain sign-in.
    me = make_user(email="me6@example.com")
    _start_link(client, "me6@example.com")

    older = _sign_in(client, google, state="an-older-flow", sub="sub-6", email="g6@example.com")
    assert older.headers["location"] == "/?oauth=cancelled"
    r = _sign_in(client, google, sub="sub-6", email="g6@example.com")

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


def test_a_connect_from_a_tab_showing_another_account_starts_nothing(client, google, make_user):
    # A navigation carries no headers, so the tab names the account it was built for in the query.
    # One still showing the account signed in before must not start a link onto the one signed in
    # now: it lands on the app, which boots as that one and says why.
    me = make_user(email="me7@example.com")
    _password_login(client, "me7@example.com")

    r = client.get(
        f"/auth/google/connect?access=identity&expect={me.id + 1}", follow_redirects=False
    )

    assert r.headers["location"] == auth_routes.ACCOUNT_CHANGED_LANDING == "/?account=changed"
    assert google.asked == {}
    assert client.get("/api/me").json()["email"] == "me7@example.com"


def test_a_connect_naming_the_signed_in_account_goes_to_google(client, google, make_user):
    me = make_user(email="me8@example.com")
    _password_login(client, "me8@example.com")

    r = client.get(f"/auth/google/connect?access=identity&expect={me.id}", follow_redirects=False)

    assert r.headers["location"] == "/fake-google"
    assert google.asked["state"]


@pytest.mark.parametrize(
    ("access", "scope", "prompt"),
    [
        ("identity", "openid email profile", "select_account"),
        ("read", auth_routes.READ_SCOPES, "consent"),
        ("write", auth_routes.WRITE_SCOPES, "consent"),
    ],
)
def test_one_connect_route_asks_google_for_what_the_access_needs(
    client, google, make_user, access, scope, prompt
):
    make_user(email=f"ask-{access}@example.com")
    _password_login(client, f"ask-{access}@example.com")

    assert _connect(client, access).headers["location"] == "/fake-google"

    assert (google.asked["scope"], google.asked["prompt"]) == (scope, prompt)
    assert google.asked["access_type"] == "offline"


@pytest.mark.parametrize("access", ["read", "write"])
def test_a_youtube_grant_attaches_to_the_password_account_and_returns_to_the_app(
    client, db, google, make_user, access
):
    me = make_user(email=f"grant-{access}@example.com")
    _password_login(client, f"grant-{access}@example.com")
    _connect(client, access)

    r = _sign_in(client, google, sub=f"sub-grant-{access}", email=f"g-{access}@example.com")

    # Back to the app, not the "linked" notice: the onboarding wizard picks up where it was.
    assert r.headers["location"] == "/"
    db.refresh(me)
    assert me.google_sub == f"sub-grant-{access}"
    assert _rows(db, User, User.google_sub, f"sub-grant-{access}") == [me]


@pytest.mark.parametrize("access", ["", "admin", "READ"])
def test_an_unknown_access_starts_no_connect(client, google, make_user, access):
    make_user(email="odd@example.com")
    _password_login(client, "odd@example.com")

    assert _connect(client, access).status_code == 422
    assert google.asked == {}


def test_a_connect_needs_a_signed_in_account(client, google):
    assert _connect(client, "identity").status_code == 401
    assert google.asked == {}


@pytest.mark.parametrize("access", ["identity", "read", "write"])
def test_without_google_configured_a_connect_says_it_failed(client, make_user, monkeypatch, access):
    monkeypatch.setattr(auth_routes, "google_oauth", lambda _db: None)
    make_user(email=f"nog-{access}@example.com")
    _password_login(client, f"nog-{access}@example.com")

    assert _connect(client, access).headers["location"] == "/?link=error"


@pytest.mark.parametrize("path", ["/auth/link", "/auth/upgrade", "/auth/upgrade?access=write"])
def test_the_old_connect_routes_are_gone(client, google, make_user, path):
    make_user(email="old@example.com")
    _password_login(client, "old@example.com")

    assert client.get(path, follow_redirects=False).status_code == 404
    assert google.asked == {}