peter
6645caa157
/auth/link and /auth/upgrade become /auth/google/connect?access=identity|read|write: one handler, one table of what each access asks Google for, the same callback markers. An unknown access is refused (422) instead of quietly meaning read, and an instance without Google answers ?link=error for a grant too, not a silent return to the app. The frontend builds the URL in lib/googleConnect, and a contract test holds it to the backend route and its access values.
340 lines
13 KiB
Python
340 lines
13 KiB
Python
"""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
|
|
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):
|
|
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 _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")
|
|
|
|
|
|
@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 == {}
|