fix(auth): keep the address the user registered with, and let it sign them in (E9.5 S3)

Two halves of one rule: the address someone typed is theirs to keep, and it must be
the one that gets them back in.

A plus-tagged or dotted Gmail is often a deliberate choice -- registering as
me+siftlode@gmail.com is how you find out later which site leaked your address. The
Google callback overwrote it with the account's primary, which can never contain a
'+', so the tag vanished on the first Google sign-in and, because sign-in matched the
stored address, took their chosen address with it. The rewrite line is older than
this epic, but S3 made it reachable here by matching adoption across spellings. It
now runs only when the mailbox GENUINELY changed, and lives in a named helper rather
than inline, so the rule has one home.

The three lookups that resolve a typed address -- password login, password reset,
verify resend -- go through find_user_by_email: exact spelling first, then the unique
canonical candidate, and None when several accounts share a mailbox rather than
guessing which one a password should open. Reset and resend previously answered a
different spelling with silence (they reply uniformly by design), which was a dead end
with no way out. Their mail now goes to the address on the account, never the one
typed into the form.
This commit is contained in:
2026-09-04 01:07:54 +02:00
parent 876cece881
commit c74487fbb3
3 changed files with 202 additions and 12 deletions
+30
View File
@@ -92,6 +92,36 @@ def is_allowed(db: Session, email: str) -> bool:
return email in settings.allowed_email_set or email in settings.admin_email_set
def find_user_by_email(db: Session, email: str) -> User | None:
"""The account someone means when they type an address at sign-in, password-reset or
verify-resend. None when there isn't exactly one answer.
Exact spelling wins. Failing that, the address is resolved to its mailbox
(utils.canonical_email) — so a password chosen at registration still works when the person types
another spelling of their own Gmail, and so the three flows that look an account up by typed
address stop being dead ends. Before this, requesting a reset for `xy@gmail.com` on an account
registered as `x.y@gmail.com` issued nothing and said nothing: the endpoint answers uniformly by
design, so the user got silence and no way in.
SEVERAL canonical candidates return None rather than a guess. That case is real — prod carries two
accounts for one mailbox — and picking one of them for a password sign-in would hand someone the
wrong account. The callers all treat None as "no such account", which is the uniform,
non-enumerating answer they already gave.
"""
if not email:
return None
email = email.strip().lower()
exact = db.execute(select(User).where(User.email == email)).scalar_one_or_none()
if exact is not None:
return exact
candidates = (
db.execute(select(User).where(User.email_canonical == canonical_email(email)))
.scalars()
.all()
)
return candidates[0] if len(candidates) == 1 else None
def get_or_create_demo_user(db: Session) -> User:
"""The one shared demo user, created on first use. No OAuth token / YouTube scope, so
has_read_scope/has_write_scope are False and every YouTube-touching path is closed to it."""
+35 -12
View File
@@ -25,6 +25,7 @@ from app.auth import (
READ_SCOPES,
WRITE_SCOPES,
current_user,
find_user_by_email,
get_or_create_demo_user,
google_enabled,
is_allowed,
@@ -350,13 +351,7 @@ async def callback(
log.info("Suspended account blocked at Google login: %s", email)
_notify_suspended(background, user.email)
return RedirectResponse(url="/?login=suspended")
# Keep the email in sync with Google, but never overwrite it with one another account already
# owns — that would hit the unique constraint and 500, wedging this user's sign-in (rare: a
# Google-side email change to a value that collides with a different local account).
if user.email != email:
clash = db.query(User).filter(User.email == email, User.id != user.id).one_or_none()
if clash is None:
user.email = email
_sync_google_email(db, user, email)
user.display_name = userinfo.get("name")
user.avatar_url = userinfo.get("picture")
# ADMIN_EMAILS (env) is the bootstrap admin list and always wins, so the configured admin can't
@@ -395,6 +390,30 @@ async def callback(
return RedirectResponse(url="/")
def _sync_google_email(db: Session, user: User, email: str) -> None:
"""Follow a genuine Google-side email change — and ONLY that.
A different SPELLING of the same mailbox is the user's own choice and must survive. People
register as `me+siftlode@gmail.com` precisely so a later leak from this site is identifiable, and
silently rewriting that to `me@gmail.com` destroys the thing it was for. It also locked them out
of the spelling they chose, because sign-in matches the stored address first. Google returns the
account's PRIMARY address, which need not be what they registered with — and can never contain a
`+` at all — so without this guard every plus-tagged registration lost its tag on the first Google
sign-in. (E9.5 S3 made that reachable by matching adoption across spellings; the rewrite itself is
older.)
Never overwrite with an address another account already owns either: that would hit the unique
constraint and 500, wedging this user's sign-in.
"""
if not email or user.email == email:
return
if canonical_email(user.email) == canonical_email(email):
return # same mailbox, their spelling — leave it alone
clash = db.query(User).filter(User.email == email, User.id != user.id).one_or_none()
if clash is None:
user.email = email
def _store_token(db: Session, user: User, token: dict) -> None:
"""Persist the OAuth grant on the user's single token row, creating it if absent. Google
only returns a refresh_token on (re)consent, so the previous one is kept otherwise.
@@ -740,10 +759,13 @@ def _resend_verify_if_eligible(email: str) -> None:
if not email_mod.email_enabled():
return
with SessionLocal() as db:
user = db.execute(select(User).where(User.email == email)).scalar_one_or_none()
user = find_user_by_email(db, email)
if user is not None and not user.email_verified and not user.is_demo:
raw = _issue_token(db, user, "verify", VERIFY_TTL)
email_mod.send_verify_email(email, f"{_app_base()}/#verify={raw}")
# To the address ON THE ACCOUNT, not the one just typed: the account's address is the one
# its owner chose to be reachable at, and a `+tag` there is usually deliberate (it tells
# them which site leaked, if one ever does). Same mailbox either way for a Gmail alias.
email_mod.send_verify_email(user.email, f"{_app_base()}/#verify={raw}")
@router.post("/password-login")
@@ -760,7 +782,7 @@ def password_login(
raise HTTPException(status_code=429, detail="Too many attempts. Try again shortly.")
email = (body.email or "").strip().lower()
password = body.password or ""
user = db.execute(select(User).where(User.email == email)).scalar_one_or_none()
user = find_user_by_email(db, email)
# Always run the KDF (against a decoy hash for an unknown / passwordless / demo account) so the
# response time is the same whether or not the email exists — no timing-based enumeration.
candidate_hash = user.password_hash if (user and not user.is_demo) else None
@@ -816,12 +838,13 @@ def _send_reset_if_eligible(email: str) -> None:
ineligible on purpose: its owner adds one from Settings (/auth/set-password) while signed in, which
proves identity through the session rather than by mail."""
with SessionLocal() as db:
user = db.execute(select(User).where(User.email == email)).scalar_one_or_none()
user = find_user_by_email(db, email)
if user is None or user.is_demo:
return
if user.password_hash or user.password_revoked_at:
raw = _issue_token(db, user, "reset", RESET_TTL)
email_mod.send_password_reset(email, f"{_app_base()}/#reset={raw}")
# The account's own address, as above — never the spelling that was typed into the form.
email_mod.send_password_reset(user.email, f"{_app_base()}/#reset={raw}")
@router.post("/password-reset/confirm")
+137
View File
@@ -156,3 +156,140 @@ def test_manually_whitelisting_another_spelling_reuses_the_row(db):
rows = db.execute(select(Invite)).scalars().all()
assert len(rows) == 1
assert rows[0].email == "hartmann.peter1986@gmail.com" # the original spelling is preserved
def _capture_reset_links(monkeypatch) -> list[str]:
"""Patch the reset mailer and collect the recipients it is asked to write to."""
from app.routes import auth as auth_mod
sent: list[str] = []
monkeypatch.setattr(auth_mod.email_mod, "send_password_reset", lambda to, link: sent.append(to))
return sent
# --- signing in and recovering an account, whichever spelling of it the person types -------------
#
# The three flows that look an account up BY TYPED ADDRESS used to match literally, so a Gmail user
# who registered `x.y@` and typed `xy@` met "Invalid email or password" at sign-in — and, worse,
# total silence from password-reset and verify-resend, which answer uniformly by design.
def test_the_password_works_with_any_spelling_of_the_same_mailbox(db):
from app.auth import find_user_by_email
make = User(
email="t.e.st+tag@gmail.com", email_verified=True, is_active=True, password_hash="x"
)
db.add(make)
db.commit()
for typed in (
"t.e.st+tag@gmail.com", # as registered
"test@gmail.com", # the mailbox itself
"t.e.st@gmail.com", # dots only
"TEST@googlemail.com", # the alias domain
):
assert find_user_by_email(db, typed) is make, typed
def test_an_unrelated_address_still_finds_nothing(db):
from app.auth import find_user_by_email
db.add(User(email="someone@gmail.com", email_verified=True, is_active=True))
db.commit()
assert find_user_by_email(db, "someoneelse@gmail.com") is None
assert find_user_by_email(db, "some.one@example.com") is None # dots matter off Gmail
def test_the_exact_spelling_wins_when_two_accounts_share_a_mailbox(db):
# Prod carries such a pair (ids 13 and 14). Whoever types their own address in full must land on
# their own account, not on whichever row happens to come first.
from app.auth import find_user_by_email
dotted = User(email="npeter83.temu@gmail.com", email_verified=True, is_active=True)
plain = User(email="npeter83temu@gmail.com", email_verified=True, is_active=True)
db.add_all([dotted, plain])
db.commit()
assert find_user_by_email(db, "npeter83.temu@gmail.com") is dotted
assert find_user_by_email(db, "npeter83temu@gmail.com") is plain
def test_an_ambiguous_spelling_refuses_rather_than_guessing(db):
# Neither spelling is exact, and both are the same mailbox. Handing a password sign-in one of
# them at random would let someone into the wrong account, so this answers "no such account" —
# the same uniform reply the caller already gives.
from app.auth import find_user_by_email
db.add(User(email="npeter83.temu@gmail.com", email_verified=True, is_active=True))
db.add(User(email="npeter83temu@gmail.com", email_verified=True, is_active=True))
db.commit()
assert find_user_by_email(db, "n.peter83.temu+x@gmail.com") is None
def test_a_reset_link_goes_to_the_address_on_the_account(db, monkeypatch):
"""Typing another spelling must still reach the owner — and the mail must go to the address they
registered with, tag and all, not to the one typed into the form."""
from app.routes.auth import _send_reset_if_eligible
db.add(
User(
email="me+siftlode@gmail.com",
email_verified=True,
is_active=True,
password_hash="x",
)
)
db.commit()
sent = _capture_reset_links(monkeypatch)
_send_reset_if_eligible("me@gmail.com")
assert sent == ["me+siftlode@gmail.com"]
# --- a Google sign-in must not rewrite the address its owner chose ------------------------------
def test_a_google_sign_in_keeps_a_plus_tagged_address(db):
"""People register as `me+siftlode@gmail.com` so a later leak from this site is identifiable.
Google returns the account's primary address, which can never contain `+`, so following it
blindly destroyed the tag on the first Google sign-in — and, since sign-in matches the stored
address first, stopped them signing in with the address they had chosen."""
from app.routes.auth import _sync_google_email
user = User(email="me+siftlode@gmail.com", email_verified=True, is_active=True)
db.add(user)
db.commit()
_sync_google_email(db, user, "me@gmail.com")
assert user.email == "me+siftlode@gmail.com"
def test_a_google_sign_in_follows_a_real_mailbox_change(db):
# The case the sync exists for: the Google account genuinely moved to a different mailbox.
from app.routes.auth import _sync_google_email
user = User(email="old.address@gmail.com", email_verified=True, is_active=True)
db.add(user)
db.commit()
_sync_google_email(db, user, "brand-new@example.com")
assert user.email == "brand-new@example.com"
def test_a_mailbox_change_is_refused_when_another_account_owns_it(db):
from app.routes.auth import _sync_google_email
db.add(User(email="taken@example.com", email_verified=True, is_active=True))
user = User(email="mover@gmail.com", email_verified=True, is_active=True)
db.add(user)
db.commit()
_sync_google_email(db, user, "taken@example.com")
assert user.email == "mover@gmail.com" # unique constraint would have 500'd the sign-in