CVE-2022-4722: Improper Authentication (CWE-287)
value: Authentication Bypass by Primary Weakness in GitHub repository ikus060/rdiffweb prior to 2.5.5.
CWE description: When an actor claims to have a given identity, the product does not prove or insufficiently proves that the claim is correct.
CVSS v3 base score: 7.2
Published: 2022-12-27T15:15Z
Repository: https://github.com/ikus060/rdiffweb
Language: Python
Files changed: README.md, rdiffweb/controller/tests/test_page_login.py, rdiffweb/controller/tests/test_page_prefs_general.py, rdiffweb/core/login.py, rdiffweb/core/model/__init__.py, rdiffweb/core/model/_user.py, rdiffweb/core/model/tests/test_user.py, rdiffweb/core/tests/test_rdw_templating.py
Commit date: 2022-12-23 14:02:50 -0500
Commit message: Make username case-insensitive

Diff: diff --git a/README.md b/README.md
index d621ee939..8d357d640 100644
--- a/README.md
+++ b/README.md
@@ -24,21 +24,21 @@ Welcome to Rdiffweb
 Rdiffweb is a web application that allows you to view repositories generated
 by [rdiff-backup](https://rdiff-backup.net/). The purpose of this
 application is to ease the management of backups and quickly restore your data
 with a rich and powerful web interface.
 
-Rdiffweb is written in Python and is released as open source project under the 
+Rdiffweb is written in Python and is released as open source project under the
 GNU GENERAL PUBLIC LICENSE (GPL). All source code and documentation are
 Copyright Rdiffweb contributors.
 
 Rdiffweb is actively developed by [IKUS Soft](https://www.ikus-soft.com/)
 since November 2014.
 
 The Rdiffweb source code is hosted on [Gitlab](https://gitlab.com/ikus-soft/rdiffweb)
 and mirrored to [Github](https://github.com/ikus060/rdiffweb).
 
-The Rdiffweb website is https://rdiffweb.org/.
+The Rdiffweb website is <https://rdiffweb.org/>.
 
 ## Features
 
 With its rich web interface Rdiffweb provide a notable list of features:
 
@@ -98,11 +98,11 @@ You should read the [Documentation](https://www.ikus-soft.com/archive/rdiffweb/d
 
 Rdiffweb users should use the [Rdiffweb mailing list](https://groups.google.com/forum/#!forum/rdiffweb).
 
 ### Bug Reports
 
-Bug reports should be reported on the Rdiffweb Gitlab at https://gitlab.com/ikus-soft/rdiffweb/-/issues
+Bug reports should be reported on the Rdiffweb Gitlab at <https://gitlab.com/ikus-soft/rdiffweb/-/issues>
 
 ### Professional support
 
 Professional support for Rdiffweb is available by contacting [IKUS Soft](https://www.ikus-soft.com/en/support/#form).
 
@@ -112,10 +112,15 @@ Professional support for Rdiffweb is available by contacting [IKUS Soft](https:/
 
 * Fix loading of Charts in Status page
 * Ensure Gmail and other mail client doesn't create hyperlink automatically for any nodification sent by Rdiffweb to avoid phishing - credit to [Nehal Pillai](https://www.linkedin.com/in/nehal-pillai-02a854172)
 * Sent email notification to user when a new SSH Key get added - credit to [Nehal Pillai](https://www.linkedin.com/in/nehal-pillai-02a854172)
 * Ratelimit "Resend code to my email" in Two-Factor Authentication view - credit to [Nehal Pillai](https://www.linkedin.com/in/nehal-pillai-02a854172)
+* Username are not case-insensitive - credits to [raiders0786](https://www.linkedin.com/in/chirag-agrawal-770488144/)
+
+Breaking changes:
+
+* Username with different cases (e.g.: admin vs Ammin) are not supported. If your database contains such username make sure to remove them before upgrading otherwise Rdiffweb will not start.
 
 ## 2.5.4 (2022-12-19)
 
 * Discard `X-Forwarded-Host` headers credit to [Anishka Shukla](https://github.com/anishkashukla)
 * Create proper symbolic link of `chartkick.js` on Ubuntu Jammy to fix loading of Charts in web interface
@@ -376,11 +381,11 @@ Maintenance release to fix minor issues
 * Define a default limit to graph statistics to make it display faster
 * Fix `get-quota-cmd` option to properly return a value
 
 ## 2.1.0 (2021-01-15)
 
-* Debian package: Remove dh-systemd from Debian build dependencies (https://bugs.debian.org/871312we)
+* Debian package: Remove dh-systemd from Debian build dependencies (<https://bugs.debian.org/871312we>)
 * Improve Quota management:
   * `QuotaSetCmd`, `QuotaGetCmd` and `QuotaUsedCmd` options could be used to customize how to set the quota for your environment.
   * Display user's quota in User View
   * Display user's quota in Admin View
   * Allow admin to update user quota from Admin View when `QuotaSetCmd` is defined.
diff --git a/rdiffweb/controller/tests/test_page_login.py b/rdiffweb/controller/tests/test_page_login.py
index 1b3d00248..997804523 100644
--- a/rdiffweb/controller/tests/test_page_login.py
+++ b/rdiffweb/controller/tests/test_page_login.py
@@ -62,10 +62,19 @@ class LoginPageTest(rdiffweb.test.WebCase):
         session = DbSession(id=self.session_id)
         session.load()
         self.assertEqual('admin', session.get(SESSION_KEY))
         self.assertIsNotNone(session.get(LOGIN_TIME))
 
+    def test_login_case_insensitive(self):
+        # When authenticating with valid credentials with all uppercase username
+        self.getPage('/login/', method='POST', body={'login': self.USERNAME.upper(), 'password': self.PASSWORD})
+        # Then a new session_id is generated
+        self.assertStatus('303 See Other')
+        self.assertHeaderItemValue('Location', self.baseurl + '/')
+        self.getPage('/')
+        self.assertStatus(200)
+
     def test_cookie_http_only(self):
         # Given a request made to rdiffweb
         # When receiving the response
         self.getPage('/')
         # Then the header contains Set-Cookie with HttpOnly
diff --git a/rdiffweb/controller/tests/test_page_prefs_general.py b/rdiffweb/controller/tests/test_page_prefs_general.py
index e7bba6669..029c168bc 100644
--- a/rdiffweb/controller/tests/test_page_prefs_general.py
+++ b/rdiffweb/controller/tests/test_page_prefs_general.py
@@ -83,12 +83,12 @@ class PagePrefGeneralTest(rdiffweb.test.WebCase):
             body={'action': 'set_profile_info', 'email': 'test@test.com', 'username': 'test'},
         )
         self.assertStatus(303)
         self.getPage(self.PREFS)
         self.assertInBody("Profile updated successfully.")
-        # Then database is updated with fullname
-        user = UserObject.query.filter(UserObject.username == self.USERNAME).first()
+        # Then database is not updated with new username.
+        user = UserObject.get_user(self.USERNAME)
         self.assertIsNotNone(user)
         self.assertEqual("test@test.com", user.email)
 
     @parameterized.expand(
         [
@@ -110,11 +110,11 @@ class PagePrefGeneralTest(rdiffweb.test.WebCase):
             self.assertStatus(303)
             self.getPage(self.PREFS)
             self.assertInBody("Profile updated successfully.")
             # Then database is updated with fullname
             self.assertInBody(new_fullname)
-            user = UserObject.query.filter(UserObject.username == self.USERNAME).first()
+            user = UserObject.get_user(self.USERNAME)
             self.assertEqual(new_fullname, user.fullname)
         else:
             self.assertStatus(200)
             self.assertNotInBody("Profile updated successfully.")
 
@@ -123,11 +123,11 @@ class PagePrefGeneralTest(rdiffweb.test.WebCase):
         # When trying to update full name using GET method
         self.getPage(self.PREFS + '?action=set_profile_info&email=test@test.com')
         # Then nothing happen
         self.assertStatus(200)
         self.assertNotInBody("Profile updated successfully.")
-        user = UserObject.query.filter(UserObject.username == self.USERNAME).first()
+        user = UserObject.get_user(self.USERNAME)
         self.assertEqual("", user.fullname)
 
     def test_change_fullname_too_long(self):
         # Given an authenticated user
         # When update the fullname
@@ -135,11 +135,11 @@ class PagePrefGeneralTest(rdiffweb.test.WebCase):
         # Then page return with error message
         self.assertStatus(200)
         self.assertNotInBody("Profile updated successfully.")
         self.assertInBody("Fullname too long.")
         # Then database is not updated
-        user = UserObject.query.filter(UserObject.username == self.USERNAME).first()
+        user = UserObject.get_user(self.USERNAME)
         self.assertEqual("", user.fullname)
 
     def test_change_email(self):
         self._set_profile_info("test@test.com")
         self.assertStatus(303)
diff --git a/rdiffweb/core/login.py b/rdiffweb/core/login.py
index 607c8e683..2fa9fea5a 100644
--- a/rdiffweb/core/login.py
+++ b/rdiffweb/core/login.py
@@ -48,11 +48,11 @@ class LoginPlugin(SimplePlugin):
 
     def authenticate(self, username, password):
         """
         Only verify the user's credentials using the database store.
         """
-        user = UserObject.query.filter_by(username=username).first()
+        user = UserObject.get_user(username)
         if user and user.validate_password(password):
             return username, {}
         return False
 
     def login(self, username, password):
@@ -67,11 +67,11 @@ class LoginPlugin(SimplePlugin):
         real_username = authenticates[0][0]
         extra_attrs = authenticates[0][1]
         fullname = extra_attrs.get('_fullname', None)
         email = extra_attrs.get('_email', None)
         # When enabled, create missing userobj in database.
-        userobj = UserObject.query.filter_by(username=username).first()
+        userobj = UserObject.get_user(username)
         if userobj is None and self.add_missing_user:
             try:
                 # At this point, we need to create a new user in database.
                 # In case default values are invalid, let evaluate them
                 # before creating the user in database.
diff --git a/rdiffweb/core/model/__init__.py b/rdiffweb/core/model/__init__.py
index ea057b06c..df13f84c1 100644
--- a/rdiffweb/core/model/__init__.py
+++ b/rdiffweb/core/model/__init__.py
@@ -13,78 +13,95 @@
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
+import logging
+import sys
+
 import cherrypy
 from sqlalchemy import event
+from sqlalchemy.exc import IntegrityError
 
 from ._repo import RepoObject  # noqa
 from ._session import DbSession, SessionObject  # noqa
 from ._sshkey import SshKey  # noqa
 from ._token import Token  # noqa
-from ._user import DuplicateSSHKeyError, UserObject  # noqa
+from ._user import DuplicateSSHKeyError, UserObject, user_username_index  # noqa
 
 Base = cherrypy.tools.db.get_base()
 
+logger = logging.getLogger(__name__)
+
+
+def _column_add(connection, column):
+    if _column_exists(connection, column):
+        return
+    table_name = column.table.fullname
+    column_name = column.name
+    column_type = column.type.compile(connection.engine.dialect)
+    connection.engine.execute('ALTER TABLE %s ADD COLUMN %s %s' % (table_name, column_name, column_type))
+
+
+def _column_exists(connection, column):
+    table_name = column.table.fullname
+    column_name = column.name
+    if 'SQLite' in connection.engine.dialect.__class__.__name__:
+        sql = "SELECT COUNT(*) FROM pragma_table_info('%s') WHERE LOWER(name)=LOWER('%s')" % (
+            table_name,
+            column_name,
+        )
+    else:
+        sql = "SELECT COUNT(*) FROM information_schema.columns WHERE table_name='%s' and column_name='%s'" % (
+            table_name,
+            column_name,
+        )
+    data = connection.engine.execute(sql).first()
+    return data[0] >= 1
+
+
+def _index_exists(connection, index_name):
+    if 'SQLite' in connection.engine.dialect.__class__.__name__:
+        sql = "SELECT name FROM sqlite_master WHERE type = 'index' AND name = '%s';" % (index_name)
+    else:
+        sql = "SELECT * FROM pg_indexes WHERE indexname = '%s'" % (index_name)
+    return connection.engine.execute(sql).first() is not None
+
 
 @event.listens_for(Base.metadata, 'after_create')
 def db_after_create(target, connection, **kw):
     """
     Called on database creation to update database schema.
     """
 
-    def exists(column):
-        table_name = column.table.fullname
-        column_name = column.name
-        if 'SQLite' in connection.engine.dialect.__class__.__name__:
-            sql = "SELECT COUNT(*) FROM pragma_table_info('%s') WHERE LOWER(name)=LOWER('%s')" % (
-                table_name,
-                column_name,
-            )
-        else:
-            sql = "SELECT COUNT(*) FROM information_schema.columns WHERE table_name='%s' and column_name='%s'" % (
-                table_name,
-                column_name,
-            )
-        data = connection.engine.execute(sql).first()
-        return data[0] >= 1
-
-    def add_column(column):
-        if exists(column):
-            return
-        table_name = column.table.fullname
-        column_name = column.name
-        column_type = column.type.compile(connection.engine.dialect)
-        connection.engine.execute('ALTER TABLE %s ADD COLUMN %s %s' % (table_name, column_name, column_type))
-
     if getattr(connection, '_transaction', None):
         connection._transaction.commit()
 
     # Add repo's Encoding
-    add_column(RepoObject.__table__.c.Encoding)
-    add_column(RepoObject.__table__.c.keepdays)
+    _column_add(connection, RepoObject.__table__.c.Encoding)
+    _column_add(connection, RepoObject.__table__.c.keepdays)
 
     # Create column for roles using "isadmin" column. Keep the
     # original column in case we need to revert to previous version.
-    if not exists(UserObject.__table__.c.role):
-        add_column(UserObject.__table__.c.role)
+    if not _column_exists(connection, UserObject.__table__.c.role):
+        _column_add(connection, UserObject.__table__.c.role)
         UserObject.query.filter(UserObject._is_admin == 1).update({UserObject.role: UserObject.ADMIN_ROLE})
 
     # Add user's fullname column
-    add_column(UserObject.__table__.c.fullname)
+    _column_add(connection, UserObject.__table__.c.fullname)
 
     # Add user's mfa column
-    add_column(UserObject.__table__.c.mfa)
+    _column_add(connection, UserObject.__table__.c.mfa)
 
     # Re-create session table if Number column is missing
-    if not exists(SessionObject.__table__.c.Number):
+    if not _column_exists(connection, SessionObject.__table__.c.Number):
         SessionObject.__table__.drop()
         SessionObject.__table__.create()
 
     if getattr(connection, '_transaction', None):
         connection._transaction.commit()
+
     # Remove preceding and leading slash (/) generated by previous
     # versions. Also rename '.' to ''
     result = RepoObject.query.all()
     for row in result:
         if row.repopath.startswith('/') or row.repopath.endswith('/'):
@@ -99,5 +116,24 @@ def db_after_create(target, connection, **kw):
     for row in result:
         if prev_repo[0] == row.userid and (prev_repo[1] == row.repopath or row.repopath.startswith(prev_repo[1] + '/')):
             row.delete()
         else:
             prev_repo = (row.userid, row.repopath)
+
+    # Fix username case insensitive unique
+    if not _index_exists(connection, 'user_username_index'):
+        duplicate_users = (
+            UserObject.query.with_entities(func.lower(UserObject.username))
+            .group_by(func.lower(UserObject.username))
+            .having(func.count(UserObject.username) > 1)
+        ).all()
+        try:
+            user_username_index.create()
+        except IntegrityError:
+            msg = (
+                'Failure to upgrade your database to make Username case insensitive. '
+                'You must downgrade and deleted duplicate Username. '
+                '%s' % '\n'.join([str(k) for k in duplicate_users]),
+            )
+            logger.error(msg)
+            print(msg, file=sys.stderr)
+            raise SystemExit(12)
diff --git a/rdiffweb/core/model/_user.py b/rdiffweb/core/model/_user.py
index 49cf0893d..9b06363c6 100644
--- a/rdiffweb/core/model/_user.py
+++ b/rdiffweb/core/model/_user.py
@@ -18,11 +18,11 @@ import logging
 import os
 import secrets
 import string
 
 import cherrypy
-from sqlalchemy import Column, Integer, SmallInteger, String, and_, event, inspect, or_
+from sqlalchemy import Column, Index, Integer, SmallInteger, String, and_, event, func, inspect, or_
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.hybrid import hybrid_property
 from sqlalchemy.orm import deferred, relationship, validates
 from zxcvbn import zxcvbn
 
@@ -72,11 +72,11 @@ class UserObject(Base):
     PATTERN_EMAIL = r"[a-z0-9._%+-]+@[a-z0-9.-]+\.[a-z]{2,4}$"
     PATTERN_FULLNAME = r"""[^!"#$%&()*+,./:;<=>?@[\]_{|}~]+$"""
     PATTERN_USERNAME = r"[a-zA-Z0-9_.\-]+$"
 
     userid = Column('UserID', Integer, primary_key=True)
-    username = Column('Username', String, nullable=False, unique=True)
+    username = Column('Username', String, nullable=False)
     hash_password = Column('Password', String, nullable=False, default="")
     user_root = Column('UserRoot', String, nullable=False, default="")
     _is_admin = deferred(
         Column(
             'IsAdmin',
@@ -108,12 +108,12 @@ class UserObject(Base):
         order_by=lambda: RepoObject.repopath,
     )
 
     @classmethod
     def get_user(cls, user):
-        """Return a user object."""
-        return UserObject.query.filter(UserObject.username == user).first()
+        """Return a user object with username case-insensitive"""
+        return UserObject.query.filter(func.lower(UserObject.username) == user.lower()).first()
 
     @classmethod
     def create_admin_user(cls, default_username, default_password):
         # Check if admin user exists. If not, created it.
         userobj = UserObject.get_user(default_username)
@@ -411,10 +411,14 @@ class UserObject(Base):
 
     def validate_password(self, password):
         return check_password(password, self.hash_password)
 
 
+# Username should be case insensitive
+user_username_index = Index('user_username_index', func.lower(UserObject.username), unique=True)
+
+
 @event.listens_for(UserObject.hash_password, "set")
 def hash_password_set(target, value, oldvalue, initiator):
     if value and value != oldvalue:
         cherrypy.engine.publish('user_password_changed', target)
 
diff --git a/rdiffweb/core/model/tests/test_user.py b/rdiffweb/core/model/tests/test_user.py
index 9044e11f3..569ac52a0 100644
--- a/rdiffweb/core/model/tests/test_user.py
+++ b/rdiffweb/core/model/tests/test_user.py
@@ -101,10 +101,20 @@ class UserObjectTest(rdiffweb.test.WebCase):
         with self.assertRaises(ValueError):
             UserObject.add_user('denise')
         # Check if listener called
         self.listener.user_added.assert_not_called()
 
+    def test_add_user_with_duplicate_caseinsensitive(self):
+        """Add user to database."""
+        user = UserObject.add_user('denise')
+        user.commit()
+        self.listener.user_added.reset_mock()
+        with self.assertRaises(ValueError):
+            UserObject.add_user('dEnIse')
+        # Check if listener called
+        self.listener.user_added.assert_not_called()
+
     def test_add_user_with_password(self):
         """Add user to database with password."""
         userobj = UserObject.add_user('jo', 'password')
         userobj.commit()
         self.assertIsNotNone(UserObject.get_user('jo'))
@@ -156,10 +166,17 @@ class UserObjectTest(rdiffweb.test.WebCase):
         self.assertEqual('broker-repo', obj.repo_objs[0].name)
         self.assertEqual(-1, obj.repo_objs[0].maxage)
         self.assertEqual('testcases', obj.repo_objs[1].name)
         self.assertEqual(3, obj.repo_objs[1].maxage)
 
+    def test_get_user_case_insensitive(self):
+        userobj1 = UserObject.get_user(self.USERNAME)
+        userobj2 = UserObject.get_user(self.USERNAME.lower())
+        userobj3 = UserObject.get_user(self.USERNAME.upper())
+        self.assertEqual(userobj1, userobj2)
+        self.assertEqual(userobj2, userobj3)
+
     def test_get_user_with_invalid_user(self):
         self.assertIsNone(UserObject.get_user('invalid'))
 
     def test_get_set(self):
         user = UserObject.add_user('larry', 'password')
diff --git a/rdiffweb/core/tests/test_rdw_templating.py b/rdiffweb/core/tests/test_rdw_templating.py
index c8d8bab3e..4dd086291 100644
--- a/rdiffweb/core/tests/test_rdw_templating.py
+++ b/rdiffweb/core/tests/test_rdw_templating.py
@@ -108,11 +108,11 @@ class ListParentsTest(WebCase):
 
 
 class UrlForTest(WebCase):
     @property
     def repo_obj(self):
-        user = UserObject.query.filter(UserObject.username == 'admin').first()
+        user = UserObject.get_user('admin')
         return RepoObject.query.filter(RepoObject.user == user, RepoObject.repopath == self.REPO).first()
 
     def test_url_for_absolute_path(self):
         self.assertEqual(cherrypy.server.base() + '/static/js/jquery.min.js', url_for('/static/js/jquery.min.js'))
 

Vulnerable code: Rdiffweb is written in Python and is released as open source project under the 
The Rdiffweb website is https://rdiffweb.org/.
Bug reports should be reported on the Rdiffweb Gitlab at https://gitlab.com/ikus-soft/rdiffweb/-/issues
* Debian package: Remove dh-systemd from Debian build dependencies (https://bugs.debian.org/871312we)
        # Then database is updated with fullname
        user = UserObject.query.filter(UserObject.username == self.USERNAME).first()
            user = UserObject.query.filter(UserObject.username == self.USERNAME).first()
        user = UserObject.query.filter(UserObject.username == self.USERNAME).first()
        user = UserObject.query.filter(UserObject.username == self.USERNAME).first()
        user = UserObject.query.filter_by(username=username).first()
        userobj = UserObject.query.filter_by(username=username).first()
from ._user import DuplicateSSHKeyError, UserObject  # noqa
    def exists(column):
        table_name = column.table.fullname
        column_name = column.name
        if 'SQLite' in connection.engine.dialect.__class__.__name__:
            sql = "SELECT COUNT(*) FROM pragma_table_info('%s') WHERE LOWER(name)=LOWER('%s')" % (
                table_name,
                column_name,
            )
        else:
            sql = "SELECT COUNT(*) FROM information_schema.columns WHERE table_name='%s' and column_name='%s'" % (
                table_name,
                column_name,
            )
        data = connection.engine.execute(sql).first()
        return data[0] >= 1

    def add_column(column):
        if exists(column):
            return
        table_name = column.table.fullname
        column_name = column.name
        column_type = column.type.compile(connection.engine.dialect)
        connection.engine.execute('ALTER TABLE %s ADD COLUMN %s %s' % (table_name, column_name, column_type))

    add_column(RepoObject.__table__.c.Encoding)
    add_column(RepoObject.__table__.c.keepdays)
    if not exists(UserObject.__table__.c.role):
        add_column(UserObject.__table__.c.role)
    add_column(UserObject.__table__.c.fullname)
    add_column(UserObject.__table__.c.mfa)
    if not exists(SessionObject.__table__.c.Number):
from sqlalchemy import Column, Integer, SmallInteger, String, and_, event, inspect, or_
    username = Column('Username', String, nullable=False, unique=True)
        """Return a user object."""
        return UserObject.query.filter(UserObject.username == user).first()
        user = UserObject.query.filter(UserObject.username == 'admin').first()
Fixed code: Rdiffweb is written in Python and is released as open source project under the
The Rdiffweb website is <https://rdiffweb.org/>.
Bug reports should be reported on the Rdiffweb Gitlab at <https://gitlab.com/ikus-soft/rdiffweb/-/issues>
* Username are not case-insensitive - credits to [raiders0786](https://www.linkedin.com/in/chirag-agrawal-770488144/)

Breaking changes:

* Username with different cases (e.g.: admin vs Ammin) are not supported. If your database contains such username make sure to remove them before upgrading otherwise Rdiffweb will not start.
* Debian package: Remove dh-systemd from Debian build dependencies (<https://bugs.debian.org/871312we>)
    def test_login_case_insensitive(self):
        # When authenticating with valid credentials with all uppercase username
        self.getPage('/login/', method='POST', body={'login': self.USERNAME.upper(), 'password': self.PASSWORD})
        # Then a new session_id is generated
        self.assertStatus('303 See Other')
        self.assertHeaderItemValue('Location', self.baseurl + '/')
        self.getPage('/')
        self.assertStatus(200)

        # Then database is not updated with new username.
        user = UserObject.get_user(self.USERNAME)
            user = UserObject.get_user(self.USERNAME)
        user = UserObject.get_user(self.USERNAME)
        user = UserObject.get_user(self.USERNAME)
        user = UserObject.get_user(username)
        userobj = UserObject.get_user(username)
import logging
import sys

from sqlalchemy.exc import IntegrityError
from ._user import DuplicateSSHKeyError, UserObject, user_username_index  # noqa
logger = logging.getLogger(__name__)


def _column_add(connection, column):
    if _column_exists(connection, column):
        return
    table_name = column.table.fullname
    column_name = column.name
    column_type = column.type.compile(connection.engine.dialect)
    connection.engine.execute('ALTER TABLE %s ADD COLUMN %s %s' % (table_name, column_name, column_type))


def _column_exists(connection, column):
    table_name = column.table.fullname
    column_name = column.name
    if 'SQLite' in connection.engine.dialect.__class__.__name__:
        sql = "SELECT COUNT(*) FROM pragma_table_info('%s') WHERE LOWER(name)=LOWER('%s')" % (
            table_name,
            column_name,
        )
    else:
        sql = "SELECT COUNT(*) FROM information_schema.columns WHERE table_name='%s' and column_name='%s'" % (
            table_name,
            column_name,
        )
    data = connection.engine.execute(sql).first()
    return data[0] >= 1


def _index_exists(connection, index_name):
    if 'SQLite' in connection.engine.dialect.__class__.__name__:
        sql = "SELECT name FROM sqlite_master WHERE type = 'index' AND name = '%s';" % (index_name)
    else:
        sql = "SELECT * FROM pg_indexes WHERE indexname = '%s'" % (index_name)
    return connection.engine.execute(sql).first() is not None

    _column_add(connection, RepoObject.__table__.c.Encoding)
    _column_add(connection, RepoObject.__table__.c.keepdays)
    if not _column_exists(connection, UserObject.__table__.c.role):
        _column_add(connection, UserObject.__table__.c.role)
    _column_add(connection, UserObject.__table__.c.fullname)
    _column_add(connection, UserObject.__table__.c.mfa)
    if not _column_exists(connection, SessionObject.__table__.c.Number):


    # Fix username case insensitive unique
    if not _index_exists(connection, 'user_username_index'):
        duplicate_users = (
            UserObject.query.with_entities(func.lower(UserObject.username))
            .group_by(func.lower(UserObject.username))
            .having(func.count(UserObject.username) > 1)
        ).all()
        try:
            user_username_index.create()
        except IntegrityError:
            msg = (
                'Failure to upgrade your database to make Username case insensitive. '
                'You must downgrade and deleted duplicate Username. '
                '%s' % '\n'.join([str(k) for k in duplicate_users]),
            )
            logger.error(msg)
            print(msg, file=sys.stderr)
            raise SystemExit(12)
from sqlalchemy import Column, Index, Integer, SmallInteger, String, and_, event, func, inspect, or_
    username = Column('Username', String, nullable=False)
        """Return a user object with username case-insensitive"""
        return UserObject.query.filter(func.lower(UserObject.username) == user.lower()).first()
# Username should be case insensitive
user_username_index = Index('user_username_index', func.lower(UserObject.username), unique=True)


    def test_add_user_with_duplicate_caseinsensitive(self):
        """Add user to database."""
        user = UserObject.add_user('denise')
        user.commit()
        self.listener.user_added.reset_mock()
        with self.assertRaises(ValueError):
            UserObject.add_user('dEnIse')
        # Check if listener called
        self.listener.user_added.assert_not_called()

    def test_get_user_case_insensitive(self):
        userobj1 = UserObject.get_user(self.USERNAME)
        userobj2 = UserObject.get_user(self.USERNAME.lower())
        userobj3 = UserObject.get_user(self.USERNAME.upper())
        self.assertEqual(userobj1, userobj2)
        self.assertEqual(userobj2, userobj3)

        user = UserObject.get_user('admin')