peter
94418a087e
A tab names the account it was claimed for on every request, beacon and upload (X-Siftlode-Expect-Account); current_user and /auth/logout refuse a mismatch with 401 account_changed and keep the session, so a tab left showing the account signed in before reloads instead of writing to the new one. /auth/login clears a connect marker left by an abandoned connect, and a link completes only on a session that still holds. The prefs merge reads the row under a lock. A tab's sessionStorage records its own owner. A page the account loses is given up, address bar and remembered page included. The panel layouts are preferences; the storage contract also refuses a key handed to the Storage API or to usePref outside its kind. A refused sign-in as someone else and a failed sign-out say so. The chromium-only e2e specs are in the gate register, the member walk waits for its requests, and devices.spec checks the save itself.
375 lines
15 KiB
Python
375 lines
15 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")
|
|
|
|
|
|
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
|
|
|
|
|
|
@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 == {}
|