Authentication Bypass by Primary Weakness in ikus060/rdiffweb

@@ -64,6 +64,15 @@ def test_login_success(self):
         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

@@ -85,8 +85,8 @@ def test_change_username_noop(self):
         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)
 
@@ -112,7 +112,7 @@ def test_change_fullname(self, new_fullname, expected_valid):
             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)
@@ -125,7 +125,7 @@ def test_change_fullname_method_get(self):
         # 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):
@@ -137,7 +137,7 @@ def test_change_fullname_too_long(self):
         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):

@@ -50,7 +50,7 @@ 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
@@ -69,7 +69,7 @@ def login(self, username, password):
         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.

@@ -15,74 +15,91 @@
 # 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()
@@ -101,3 +118,22 @@ def add_column(column):
             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)

@@ -103,6 +103,16 @@ def test_add_user_with_duplicate(self):
         # 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')
@@ -158,6 +168,13 @@ def test_get_user(self):
         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'))
 

@@ -20,7 +20,7 @@
 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
@@ -74,7 +74,7 @@ class UserObject(Base):
     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(
@@ -110,8 +110,8 @@ class UserObject(Base):
 
     @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):
@@ -413,6 +413,10 @@ 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:

@@ -110,7 +110,7 @@ def test_list_parents_with_root_subdir(self):
 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):

@@ -26,7 +26,7 @@ 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.
 
@@ -36,7 +36,7 @@ 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
 
@@ -100,7 +100,7 @@ Rdiffweb users should use the [Rdiffweb mailing list](https://groups.google.com/
 
 ### 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
 
@@ -114,6 +114,11 @@ Professional support for Rdiffweb is available by contacting [IKUS Soft](https:/
 * 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)
 
@@ -378,7 +383,7 @@ Maintenance release to fix minor issues
 
 ## 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