fix: make email lookup case-insensitive in get_user_by_email (#397)

* fix: make email lookup case-insensitive in get_user_by_email

Email addresses are case-insensitive in practice, but get_user_by_email
compared with an exact `UserModel.email == email` predicate. A user who
signed up as "User@example.com" could not be found when logging in as
"user@example.com" (and vice-versa), so the same person could fail to log
in — or be treated as a brand-new account — depending only on how their
client capitalized the address.

Compare on `func.lower(UserModel.email) == func.lower(email)` so lookups
are robust to capitalization. Minimal and backwards-compatible: it works
with existing mixed-case rows immediately, with no migration required.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* fix: enforce case-insensitive user emails

---------

Co-authored-by: developer603 <vrramsolutions@gmail.com>
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Co-authored-by: Abhishek Kumar <abhishek@a6k.me>
This commit is contained in:
developer603 2026-06-02 13:43:20 +05:30 committed by GitHub
parent 8b9059fbe2
commit acc2ef9e96
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
4 changed files with 78 additions and 5 deletions

View file

@ -0,0 +1,32 @@
"""make email case insensitive
Revision ID: 384be6596b36
Revises: 6bd9f67ec994
Create Date: 2026-06-02 07:58:00.002359
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = '384be6596b36'
down_revision: Union[str, None] = '6bd9f67ec994'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.drop_index(op.f('ix_users_email'), table_name='users')
op.create_index('ix_users_email_lower', 'users', [sa.literal_column('lower(email)')], unique=True, postgresql_where=sa.text('email IS NOT NULL'))
# ### end Alembic commands ###
def downgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.drop_index('ix_users_email_lower', table_name='users', postgresql_where=sa.text('email IS NOT NULL'))
op.create_index(op.f('ix_users_email'), 'users', ['email'], unique=True)
# ### end Alembic commands ###

View file

@ -17,6 +17,7 @@ from sqlalchemy import (
Text,
UniqueConstraint,
and_,
func,
text,
)
from sqlalchemy.orm import declarative_base, relationship
@ -67,9 +68,18 @@ class UserModel(Base):
back_populates="users",
)
is_superuser = Column(Boolean, default=False)
email = Column(String, unique=True, index=True, nullable=True)
email = Column(String, nullable=True)
password_hash = Column(String, nullable=True)
__table_args__ = (
Index(
"ix_users_email_lower",
func.lower(email),
unique=True,
postgresql_where=text("email IS NOT NULL"),
),
)
class UserConfigurationModel(Base):
__tablename__ = "user_configurations"

View file

@ -3,6 +3,7 @@ from datetime import datetime, timezone
from loguru import logger
from pydantic import ValidationError
from sqlalchemy import func
from sqlalchemy.future import select
from api.db.base_client import BaseDBClient
@ -161,15 +162,26 @@ class UserClient(BaseDBClient):
async with self.async_session() as session:
from sqlalchemy import update
stmt = update(UserModel).where(UserModel.id == user_id).values(email=email)
stmt = (
update(UserModel)
.where(UserModel.id == user_id)
.values(email=email.lower())
)
await session.execute(stmt)
await session.commit()
async def get_user_by_email(self, email: str) -> UserModel | None:
"""Fetch a user by their email address."""
"""Fetch a user by their email address (case-insensitive).
Email addresses are case-insensitive in practice, so a user who
signed up as "User@example.com" must still be found when they later
log in as "user@example.com". Compare on lower(email) so lookups are
robust to capitalization differences across sign-in flows.
"""
normalized_email = email.lower()
async with self.async_session() as session:
result = await session.execute(
select(UserModel).where(UserModel.email == email)
select(UserModel).where(func.lower(UserModel.email) == normalized_email)
)
return result.scalars().first()
@ -180,7 +192,7 @@ class UserClient(BaseDBClient):
async with self.async_session() as session:
user = UserModel(
provider_id=f"oss_{int(datetime.now(timezone.utc).timestamp())}_{uuid.uuid4()}",
email=email,
email=email.lower(),
password_hash=password_hash,
)
session.add(user)

View file

@ -0,0 +1,19 @@
import pytest
@pytest.mark.asyncio
async def test_user_email_writes_lowercase_and_looks_up_case_insensitively(
db_session,
):
user = await db_session.create_user_with_email(
email="User@Example.COM",
password_hash="hashed-password",
)
assert user.email == "user@example.com"
fetched = await db_session.get_user_by_email("USER@example.com")
assert fetched is not None
assert fetched.id == user.id
assert fetched.email == "user@example.com"