add unique index to enforce no duplicate on case-insensitive user emails

CHANGES.rst

@@ -11,7 +11,11 @@ Changes
 
 Features / Changes
 ~~~~~~~~~~~~~~~~~~~~~
-* Nothing new for the moment.
+* | Add database unique index to ensure case-insensitive ``User`` email cannot be stored.
+ |
+ | **IMPORTANT**:
+ | If any ``User`` entries with duplicate case-insensitive emails are present in the database, the application
+ will fail when performing the database migration. Resolve those cases manually before starting `Magpie`.
 
 Bug Fixes
 ~~~~~~~~~~~~~~~~~~~~~

magpie/alembic/script.py.mako

@@ -8,11 +8,12 @@ Create Date: ${create_date}
 
 import sqlalchemy as sa
 from alembic import op
+from sqlalchemy.orm.session import sessionmaker
 
 # Revision identifiers, used by Alembic.
 # pylint: disable=C0103,invalid-name # revision control variables not uppercase
-%if revision:
-revision = "${revision}"
+%if up_revision:
+revision = "${up_revision}"
 %else:
 revision = None
 %endif
@@ -32,6 +33,8 @@ depends_on = "${depends_on}"
 depends_on = None
 %endif
 
+Session = sessionmaker()
+
 
 def upgrade():
  ${upgrades if upgrades else "pass"}

magpie/alembic/versions/2022-09-01_5e5acc33adce_case_insensitive_email_constraint.py

@@ -0,0 +1,30 @@
+"""
+Case Insensitive Email Constraint
+
+Revision ID: 5e5acc33adce
+Revises: 0c6269f410cd
+Create Date: 2022-09-01 21:16:40.175730
+"""
+
+from alembic import op
+from sqlalchemy import text
+
+# Revision identifiers, used by Alembic.
+# pylint: disable=C0103,invalid-name # revision control variables not uppercase
+revision = "5e5acc33adce"
+down_revision = "0c6269f410cd"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+ op.create_index(
+ "ix_users_email_unique_case_insensitive",
+ "users",
+ [text("lower(email)")],
+ unique=True,
+ )
+
+
+def downgrade():
+ op.drop_index("ix_users_email_unique_case_insensitive", "users")

magpie/models.py

@@ -10,6 +10,7 @@
 from sqlalchemy.ext.declarative import declarative_base, declared_attr
 from sqlalchemy.ext.hybrid import hybrid_property
 from sqlalchemy.orm import relationship
+from sqlalchemy.sql import func
 from ziggurat_foundations import ziggurat_model_init
 from ziggurat_foundations.models.base import BaseModel, get_db_session
 from ziggurat_foundations.models.external_identity import ExternalIdentityMixin
@@ -512,16 +513,16 @@ def by_name_or_email(cls, user_name, email, status=None, db_session=None):
  if status is not None:
  status = [int(status) for status in status]
  query = query.in_(status)
- return query.filter((User.user_name == user_name) | (User.email == email)).first()
+ return query.filter((User.user_name == user_name) | (func.lower(User.email) == email.lower())).first()
  user = cls.by_user_name(user_name=user_name, status=status, db_session=db_session)
  if user is not None:
  return user
  if status is UserStatuses.Pending:
- return db_session.query(UserPending).filter(UserPending.email == email).first()
+ return db_session.query(UserPending).filter(func.lower(UserPending.email) == email.lower()).first()
  user = super(UserSearchService, cls).by_email(email=email, db_session=db_session)
  if user is not None and UserStatuses.get(user.status) in status:
  return user
- return db_session.query(UserPending).filter(UserPending.email == email).first()
+ return db_session.query(UserPending).filter(func.lower(UserPending.email) == email.lower()).first()
 
 
 class ExternalIdentity(ExternalIdentityMixin, Base):

magpie/ui/utils.py

@@ -558,7 +558,7 @@ def create_user(self, data):
 
  # soft pre-checks
  user_details = self.get_user_details(status="all", cookies=admin_cookies)
- if user_email in [usr["email"].lower() for usr in user_details]:
+ if (user_email or "").lower() in [usr["email"].lower() for usr in user_details]:
  data["invalid_user_email"] = True
  data["reason_user_email"] = "Conflict"
  if user_email == "":